# 开启和查询RocketMQ消息轨迹 消息轨迹是指一条消息从生产者发出到消费者消费处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。开启消息轨迹后，当消息收发不符合预期（如消息未发送成功或未消费成功）时，可以通过Message ID、Message Key或Topic的时间范围查询相关消息的消息轨迹，找到消息的实际收发状态，帮助您快速诊断消息问题。 #### 消息轨迹数据 RocketMQ系统中，一条消息的完整链路包含生产者、服务端、消费者三个角色，每个角色处理消息的过程中都会在轨迹链路中增加相关的信息，将这些信息汇聚即可获取任意消息当前的状态。 图1消息轨迹数据 ![](https://support.huaweicloud.com/usermanual-hrm/zh-cn_image_0000002321856837.png "点击放大") #### 前提条件 生产者Java客户端版本在4.9.0以上才支持事务消息轨迹，而开启SSL的RocketMQ实例，生产者和消费者的Java客户端都需在4.9.2以上版本才支持消息轨迹，若版本不满足要求，请先升级。客户端版本的查看方法请参考[查看RocketMQ消费组详情](https://support.huaweicloud.com/usermanual-hrm/hrm-ug-014.html)。 #### 开启RocketMQ消息轨迹 您可以根据实际需求，选择使用Java或Go编程语言，或基于Spring框架（Java开发）开启消息轨迹功能。 - [开启RocketMQ消息轨迹（Java）] 在客户端开启消息轨迹的方法如下： - 生产者开启消息轨迹（**除事务消息以外的消息类型** ） 构造函数的"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); ``` - [开启RocketMQ消息轨迹（Go）] 在客户端开启消息轨迹的方法如下： 1. 执行以下命令，检查是否已安装Go。 ``` go version ``` 返回如下回显时，说明Go已经安装。如果未安装Go，请[下载并安装](https://go.dev/dl/go1.16.5.linux-amd64.tar.gz)。 ``` [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管理"中获取。 | - [开启RocketMQ消息轨迹（Spring）] 在客户端开启消息轨迹的方法如下： - 生产者开启消息轨迹 在"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 { ... } ``` 如上示例代码中加粗取值的说明及获取方式请参考[表1]。 更多配置说明，请参考[消息轨迹](https://github.com/apache/rocketmq-spring/wiki/消息轨迹)，如需配置"accessChannel"，请使用默认值LOCAL。 #### 查询消息轨迹 1. 登录[RocketMQ实例控制台](https://console.huaweicloud.com/dms/?engine=reliability&locale=zh-cn#/queue/manager/instances)。 2. 单击RocketMQ实例的名称，进入实例概览页面。 3. 在左侧导航栏，单击"实例管理 \> 消息查询"，进入"消息查询"页面。 4. 选择以下任意一种方法，查询消息。 - 按Topic查询："Topic"选择待查询消息的Topic名称，"队列"选择待查询消息的队列（仅RocketMQ实例4.8.0版本需要选择），"存储时间"选择待查询消息的时间段，单击"查询"。 图2按Topic查询消息 ![](https://support.huaweicloud.com/usermanual-hrm/zh-cn_image_0000002469129169.png "点击放大") - 按Message ID查询："Topic"选择待查询消息所在的Topic名称，"Message ID"输入待查询消息的Message ID，单击"查询"。 图3按Message ID查询消息 ![](https://support.huaweicloud.com/usermanual-hrm/zh-cn_image_0000002469325577.png "点击放大") - 按Message Key查询："Topic"选择待查询消息所在的Topic名称，"Message Key"输入待查询消息的Message Key，单击"查询"。 ![](https://support.huaweicloud.com/usermanual-hrm/public_sys-resources/caution_3.0-zh-cn.png) - 按Message Key查询的条件是您在生产消息时已设置Message Key属性，建议您尽可能保证设置的Message Key是唯一的。 - 按Message Key查询时，不同版本和规格查到的最多消息条数会不同，可能会导致消息漏查。 图4按Message Key查询消息 ![](https://support.huaweicloud.com/usermanual-hrm/zh-cn_image_0000002515088768.png "点击放大") 5. 在待查询消息所在行，单击"消息轨迹"。 6. 查看消息的轨迹，确定是否生产/消费成功，如[图5]所示。 图5查看消息轨迹详情 ![](https://support.huaweicloud.com/usermanual-hrm/zh-cn_image_0000002373513389.png "点击放大") 消息轨迹的参数说明如[表2]所示。 表2消息轨迹的参数说明 | 参数 | 参数说明 | |:---|:---| | 生产者状态 | 生产者状态如下： - 发送成功：消息发送成功，服务端已经成功存储消息。 - 提交成功：允许消费者消费此事务消息。 - 回滚：事务消息将被丢弃，不允许消费者消费此事务消息。 - 未知，待确认：事务消息状态暂时无法确定，等待固定时间后，服务端向生产者进行消息回查。 | | 生产耗时 | 生产者发送消息的耗时。 单位：毫秒 | | 生产地址 | 生产者的IP地址。 | | 消费者状态 | 消费者的状态。 该参数的取值不固定，是客户根据客户端的实际消费情况设置的返回值。以下是一些常见的示例取值： - 消费成功 - 消费超时 - 消费失败 | | 消费时间 | 消费消息的时间。 时间格式为：YYYY/MM/DD hh:mm:ss | | 消费耗时 | 消费者消费消息的耗时。 单位：毫秒 | | 消费地址 | 消费者的IP地址。 | #### 相关文档 - 查询消息轨迹也可以通过调用API完成，具体请参见[查询消息轨迹](https://support.huaweicloud.com/api-hrm/ListMessageTrace.html)。 - 查询不到消息轨迹时，请参见[为什么查询不到消息轨迹](https://support.huaweicloud.com/hrm-faq/hrm_faq_033.html)。