go get github.com/apache/rocketmq-client-go/v2

// Topic name.
var topicName = "topic1"
// Construct the message content.
msg := &primitive.Message{
    Topic: topicName, // Set the topic name.
    Body:  []byte("Hello RocketMQ Go Client! This is a new message."),
}
// Set the tag.
msg.WithTag("TAG")
// Set the key.
msg.WithKeys([]string{"yourKey"})
// Send messages.
res, err := p.SendSync(context.Background(), msg)

if err != nil {
    fmt.Printf("send message error: %s\\n", err)
} else {
    fmt.Printf("send message success: result=%s\\n", res.String())
}

// Close the producer.
err = p.Shutdown()
if err != nil {
    fmt.Printf("shutdown producer error: %s", err.Error())
}

// Service access address. (Note: You need to add http:// or https:// before the access address; otherwise, it cannot be resolved.)
var serverAddress = "https://rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:9876"
// Authorized role name.
var secretKey = "admin"
// Authorized role token.
var accessKey = "eyJrZXlJZC...."
// Full namespace name.
var nameSpace = "rocketmq-xxx|namespace_go"
// Producer group name.
var groupName = "group11"
// Create a consumer.
c, err := rocketmq.NewPushConsumer(
    // Set the consumer group.
    consumer.WithGroupName(groupName),
    // Set the service address.
    consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{serverAddress})),
    // Set the ACL permissions.
    consumer.WithCredentials(primitive.Credentials{
        SecretKey: secretKey,
        AccessKey: accessKey,
    }),
    // Set the namespace name.
    consumer.WithNamespace(nameSpace),
    // Set consumption from the starting position.
    consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
    // Set the consumption mode (clustering mode by default).
    consumer.WithConsumerModel(consumer.Clustering),
    
    //For broadcasting consumption, set the instance name. Set it to the system name of the application; otherwise, pid is used, which causes repeated consumption after a restart.
    consumer.WithInstance("xxxx"),
)
if err != nil {
    fmt.Println("init consumer2 error: " + err.Error())
    os.Exit(0)
}

// Topic name.
var topicName = "topic1"
// Set the tag for subscribed messages.
selector := consumer.MessageSelector{
    Type:       consumer.TAG,
    Expression: "TagA || TagC",
}
// Set the reconsumption delay level. 18 delay levels are supported. Below is the mapping between delay levels and delay times.
// 1   2   3    4    5   6   7   8   9   10  11  12  13  14   15   16   17  18
// 1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h
delayLevel := 1
err = c.Subscribe(topicName, selector, func(ctx context.Context,
                                                              msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    fmt.Printf("subscribe callback len: %d \\n", len(msgs))
    // Set the delay level for the next consumption.
    concurrentCtx, _ := primitive.GetConcurrentlyCtx(ctx)
    concurrentCtx.DelayLevelWhenNextConsume = delayLevel // only run when return consumer.ConsumeRetryLater

    for _, msg := range msgs {
        // Simulate successful consumption after 3 retries.
        if msg.ReconsumeTimes > 3 {
            fmt.Printf("msg ReconsumeTimes > 3. msg: %v", msg)
            return consumer.ConsumeSuccess, nil
        } else {
            fmt.Printf("subscribe callback: %v \\n", msg)
        }
    }
    // Simulate consumption failure and return consumption retry to the server.
    return consumer.ConsumeRetryLater, nil
})
if err != nil {
    fmt.Println(err.Error())
}

// Start consumption.
err = c.Start()
if err != nil {
    fmt.Println(err.Error())
    os.Exit(-1)
}
time.Sleep(time.Hour)
// Release resources.
err = c.Shutdown()
if err != nil {
    fmt.Printf("shundown Consumer error: %s", err.Error())
}