消息轨迹

1. 消息轨迹数据关键属性

Producer端Consumer端Broker端
生产实例信息消费实例信息 消息的Topic
发送消息时间投递时间,投递轮次 消息存储位置
消息是否发送成功消息是否消费成功消息的Key值
发送耗时消费耗时 消息的Tag值

2. 支持消息轨迹集群部署

2.1 Broker端配置文件

这里贴出Broker端开启消息轨迹特性的properties配置文件内容：

  1. brokerClusterName=DefaultCluster
  2. brokerName=broker-a
  3. brokerId=0
  4. deleteWhen=04
  5. fileReservedTime=48
  6. brokerRole=ASYNC_MASTER
  7. flushDiskType=ASYNC_FLUSH
  8. storePathRootDir=/data/rocketmq/rootdir-a-m
  9. storePathCommitLog=/data/rocketmq/commitlog-a-m
  10. autoCreateSubscriptionGroup=true
  11. ## if msg tracing is open,the flag will be true
  12. traceTopicEnable=true
  13. listenPort=10911
  14. brokerIP1=XX.XX.XX.XX1
  15. namesrvAddr=XX.XX.XX.XX:9876

2.2 普通模式

RocketMQ集群中每一个Broker节点均用于存储Client端收集并发送过来的消息轨迹数据。因此，对于RocketMQ集群中的Broker节点数量并无要求和限制。

2.3 物理IO隔离模式

对于消息轨迹数据量较大的场景，可以在RocketMQ集群中选择其中一个Broker节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理IO完全隔离，互不影响。在该模式下，RockeMQ集群中至少有两个Broker节点，其中一个Broker节点定义为存储消息轨迹数据的服务端。

2.4 启动开启消息轨迹的Broker

nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &

3. 保存消息轨迹的Topic定义

RocketMQ的消息轨迹特性支持两种存储轨迹数据的方式：

3.1 系统级的TraceTopic

在默认情况下，消息轨迹数据是存储于系统级的TraceTopic中(其名称为：RMQ_SYS_TRACE_TOPIC)。该Topic在Broker节点启动时，会自动创建出来（如上所叙，需要在Broker端的配置文件中将traceTopicEnable的开关变量设置为true）。

3.2 用户自定义的TraceTopic

如果用户不准备将消息轨迹的数据存储于系统级的默认TraceTopic，也可以自己定义并创建用户级的Topic来保存轨迹（即为创建普通的Topic用于保存消息轨迹数据）。下面一节会介绍Client客户端的接口如何支持用户自定义的TraceTopic。

4. 支持消息轨迹的Client客户端实践

为了尽可能地减少用户业务系统使用RocketMQ消息轨迹特性的改造工作量，作者在设计时候采用对原来接口增加一个开关参数(enableMsgTrace)来实现消息轨迹是否开启；并新增一个自定义参(customizedTraceTopic)数来实现用户存储消息轨迹数据至自己创建的用户级Topic。

4.1 发送消息时开启消息轨迹

  1.         DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true);
  2.         producer.setNamesrvAddr("XX.XX.XX.XX1");
  3.         producer.start();
  4.             try {
  5.                 {
  6.                     Message msg = new Message("TopicTest",
  7.                         "TagA",
  8.                         "OrderID188",
  9.                         "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
  10.                     SendResult sendResult = producer.send(msg);
  11.                     System.out.printf("%s%n", sendResult);
  12.                 }
  14.             } catch (Exception e) {
  15.                 e.printStackTrace();
  16.             }

4.2 订阅消息时开启消息轨迹

  1.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true);
  2.         consumer.subscribe("TopicTest", "*");
  3.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  4.         consumer.setConsumeTimestamp("20181109221800");
  5.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  6.             @Override
  7.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  8.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
  9.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  10.             }
  11.         });
  12.         consumer.start();
  13.         System.out.printf("Consumer Started.%n");

4.3 支持自定义存储消息轨迹Topic

在上面的发送和订阅消息时候分别将DefaultMQProducer和DefaultMQPushConsumer实例的初始化修改为如下即可支持自定义存储消息轨迹Topic。

  1.         ##其中Topic_test11111需要用户自己预先创建，来保存消息轨迹；
  2.         DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true,"Topic_test11111");
  3.         ......
  5.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true,"Topic_test11111");
  6.         ......