tencent cloud

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

// Topic name
   var topicName = "topic1"
   // Construct 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 the message
   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())
   }

// Service access address (Note: http:// or https:// must be appended before the access address. Otherwise, it cannot be parsed.)
   var serverAddress = "http://rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:8080"
   // Authorize the role name
   var secretKey = "admin"
   // Authorize the key for the role
   var accessKey = "eyJrZXlJZC...."
   // 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 ACL permissions
       consumer.WithCredentials(primitive.Credentials{
           SecretKey: secretKey,
           AccessKey: accessKey,
       }),
       // Set consumption from the start offset
       consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
       // Set the consumption mode (cluster mode by default)
       consumer.WithConsumerModel(consumer.Clustering),

       // For broadcasting consumption, set the instance name to the system name of the application. If it is not set, the PID will be used, which can cause duplicate consumption upon 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 of messages that are subscribed to
   selector := consumer.MessageSelector{
       Type:       consumer.TAG,
       Expression: "TagA || TagC",
   }
   // Define the delay level for retrying consumption. There are 18 delay levels in total. The following is the relationship between delay levels and delay time.
   // 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 a successful consumption after three 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 a consumption failure and respond with a retry
       return consumer.ConsumeRetryLater, nil
   })
   if err != nil {
       fmt.Println(err.Error())
   }

// Initiate 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())
   }