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

// topic名称
var topicName = "topic1"
// 构造消息内容
msg := &primitive.Message{
    Topic: topicName, // 设置topic名称
    Body:  []byte("Hello RocketMQ Go Client! This is a new message."),
}
// 设置tag
msg.WithTag("TAG")
// 设置key
msg.WithKeys([]string{"yourKey"})
// 发送消息
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())
}

// 关闭生产者
err = p.Shutdown()
if err != nil {
    fmt.Printf("shutdown producer error: %s", err.Error())
}

// 服务接入地址  (注意：需要在接入地址前面加上  http:// 或 https:// 否则无法解析)
var serverAddress = "https://rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:9876"
// 授权角色名
var secretKey = "admin"
// 授权角色密钥
var accessKey = "eyJrZXlJZC...."
// 命名空间全称
var nameSpace = "rocketmq-xxx|namespace_go"
// 生产者组名称
var groupName = "group11"
// 创建consumer
c, err := rocketmq.NewPushConsumer(
    // 设置消费者组
    consumer.WithGroupName(groupName),
    // 设置服务地址
    consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{serverAddress})),
    // 设置acl权限
    consumer.WithCredentials(primitive.Credentials{
        SecretKey: secretKey,
        AccessKey: accessKey,
    }),
    // 设置命名空间名称
    consumer.WithNamespace(nameSpace),
    // 设置从起始位置开始消费
    consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
    // 设置消费模式（默认集群模式）
    consumer.WithConsumerModel(consumer.Clustering),
    
    //广播消费,设置一下实例名，设置为应用的系统名即可。如果不设置，会使用pid，这会导致重启消费重复
    consumer.WithInstance("xxxx"),
)
if err != nil {
    fmt.Println("init consumer2 error: " + err.Error())
    os.Exit(0)
}

// topic名称
var topicName = "topic1"
// 设置订阅消息的tag
selector := consumer.MessageSelector{
    Type:       consumer.TAG,
    Expression: "TagA || TagC",
}
// 设置重新消费的延迟级别，共支持18种延迟级别。下面是延迟级别与延迟时间的对应关系
// 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))
    // 设置下次消费的延迟级别
    concurrentCtx, _ := primitive.GetConcurrentlyCtx(ctx)
    concurrentCtx.DelayLevelWhenNextConsume = delayLevel // only run when return consumer.ConsumeRetryLater

    for _, msg := range msgs {
        // 模拟重试3次后消费成功
        if msg.ReconsumeTimes > 3 {
            fmt.Printf("msg ReconsumeTimes > 3. msg: %v", msg)
            return consumer.ConsumeSuccess, nil
        } else {
            fmt.Printf("subscribe callback: %v \\n", msg)
        }
    }
    // 模拟消费失败，回复重试
    return consumer.ConsumeRetryLater, nil
})
if err != nil {
    fmt.Println(err.Error())
}

// 开始消费
err = c.Start()
if err != nil {
    fmt.Println(err.Error())
    os.Exit(-1)
}
time.Sleep(time.Hour)
// 资源释放
err = c.Shutdown()
if err != nil {
    fmt.Printf("shundown Consumer error: %s", err.Error())
}