开启和查询RocketMQ消息轨迹

在客户端开启消息轨迹的方法如下：

• 生产者开启消息轨迹（除事务消息以外的消息类型）

  构造函数的“enableMsgTrace”参数传入“true”，例如：

  DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName", true);

• 生产者开启消息轨迹（事务消息）

  构造函数的“enableMsgTrace”参数传入“true”，例如：

  TransactionMQProducer producer = new TransactionMQProducer(null, "ProducerGroupName", null, true, null);

• 消费者开启消息轨迹

  构造函数的“enableMsgTrace”参数传入“true”，例如：

  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName", true);

在客户端开启消息轨迹的方法如下：

1. 执行以下命令，检查是否已安装Go。

   go version

   返回如下回显时，说明Go已经安装。如果未安装Go，请下载并安装。

   [root@ecs-test sarama]# go version
   go version go1.16.5 linux/amd64

2. 进入Go脚本所在的bin目录下。
3. 执行“touch go.mod”命令新建一个“go.mod”，并增加以下代码，添加依赖。

   module rocketmq-example-go
   go 1.13
   require (
   	github.com/apache/rocketmq-client-go/v2 v2.1.0
   )

4. 生产者开启消息轨迹（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"time"
   
   	"github.com/apache/rocketmq-client-go/v2"
   	"github.com/apache/rocketmq-client-go/v2/primitive"
   	"github.com/apache/rocketmq-client-go/v2/producer"
   )
   
   func main() {
   	namesrvs := []string{"192.168.0.1:8100"}
   	traceCfg := &primitive.TraceConfig{
   		Access:   primitive.Local,
   		Resolver: primitive.NewPassthroughResolver(namesrvs),
   	}
   
   	p, _ := rocketmq.NewProducer(
   		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
   		producer.WithRetry(2),
   		producer.WithTrace(traceCfg))   // 增加此行代码表示开启了消息轨迹。
   	err := p.Start()
   	if err != nil {
   		fmt.Printf("start producer error: %s", err.Error())
   		os.Exit(1)
   	}
   	res, err := p.SendSync(context.Background(), primitive.NewMessage("topic1",
   		[]byte("Hello RocketMQ Go Client!")))
   
   	if err != nil {
   		fmt.Printf("send message error: %s\n", err)
   	} else {
   		fmt.Printf("send message success: result=%s\n", res.String())
   	}
   
   	time.Sleep(10 * time.Second)
   
   	err = p.Shutdown()
   	if err != nil {
   		fmt.Printf("shutdown producer error: %s", err.Error())
   	}
   }

   如上代码中，192.168.0.1:8100表示RocketMQ实例的连接地址和端口，实际取值可在RocketMQ控制台的概览 > 连接信息中获取。

5. 消费者开启消息轨迹（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"time"
   
   	"github.com/apache/rocketmq-client-go/v2"
   	"github.com/apache/rocketmq-client-go/v2/consumer"
   	"github.com/apache/rocketmq-client-go/v2/primitive"
   )
   
   func main() {
   	namesrvs := []string{"192.168.0.1:8100"}
   	traceCfg := &primitive.TraceConfig{
   		Access:   primitive.Local,
   		Resolver: primitive.NewPassthroughResolver(namesrvs),
   	}
   
   	c, _ := rocketmq.NewPushConsumer(
   		consumer.WithGroupName("testGroup"),
   		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
   		consumer.WithTrace(traceCfg),   // 增加此行代码表示开启了消息轨迹。
   	)
   	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
   		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
   		fmt.Printf("subscribe callback: %v \n", msgs)
   		return consumer.ConsumeSuccess, nil
   	})
   	if err != nil {
   		fmt.Println(err.Error())
   	}
   	// Note: start after subscribe
   	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("shutdown Consumer error: %s", err.Error())
   	}
   }

   如上示例代码中加粗取值的说明及获取方式见表1。

   表1 加粗取值的说明及获取方式

   取值	取值说明	取值获取方式
   192.168.0.1:8100	表示RocketMQ实例的连接地址和端口。	在RocketMQ控制台的“概览 > 连接信息”中获取。
   testGroup	表示RocketMQ实例下创建的消费组名称。	在RocketMQ控制台的“实例管理 > 消费组管理”中获取。
   TopicTest	表示RocketMQ实例下创建的Topic名称。	在RocketMQ控制台的“实例管理 > Topic管理”中获取。

在客户端开启消息轨迹的方法如下：

• 生产者开启消息轨迹

  在“application.properties”配置文件中增加如下内容：

  rocketmq.producer.enable-msg-trace=true

• 消费者开启消息轨迹

  将“enableMsgTrace”参数值配置为“true”，例如：

  @Service
  @RocketMQMessageListener(
      topic = "test-topic-1", 
      consumerGroup = "my-consumer_test-topic-1",
      enableMsgTrace = true
  )
  public class MyConsumer implements RocketMQListener<String> {
      ...
  }

  如上示例代码中加粗取值的说明及获取方式请参考表1。

更多配置说明，请参考消息轨迹，如需配置“accessChannel”，请使用默认值LOCAL。