tencent cloud

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 = "http://rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:8080"
   // 授权角色名
   var secretKey = "admin"
   // 授权角色密钥
   var accessKey = "eyJrZXlJZC...."
   // 生产者组名称
   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.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())
   }