RocketMQ源码之消息轨迹

点击上方蓝色字关注我们~

消息轨迹在不少情况是很重要的，比如消息到底有没有发送出去，消息存储在哪个Broker，消息被哪些消费者消费、消费时间、消费耗时等等。

对于我们排查消息问题还是非常重要的。

RocketMQ 4.4.0版本开始支持消息轨迹，目前(2019-09-08)最新release版本是4.5.2。

下图是使用RocketMQ-console查看消息轨迹的数据：

下面就来看RocketMQ是如何实现消息轨迹这个新特性：

消息轨迹，必然需要对象实体存储相关数据的数据结构，比如生产者IP，发送时间、消息信息等等。

从RocketMQ源码里找到其数据结构TraceBean：

public class TraceBean {
    private static final String LOCAL_ADDRESS = UtilAll.ipToIPv4Str(UtilAll.getIP());
    private String topic = ""; // 消息topic
    private String msgId = ""; // 消息Id
    private String offsetMsgId = ""; // 消息偏移量
    private String tags = ""; // Tags用于过滤
    private String keys = ""; // Keys用于索引
    private String storeHost = LOCAL_ADDRESS; // 存储地址，默认当前机器
    private String clientHost = LOCAL_ADDRESS; // 客户端地址，在生产端就是生产者地址，默认当前机器
    private long storeTime; // 存储时间
    private int retryTimes; // 重试次数
    private int bodyLength; // 消息体长度
    private MessageType msgType; // 消息类型
    ...
}

消息轨迹上下文TraceContext：
public class TraceContext implements Comparable<TraceContext> {
    private TraceType traceType; // 追踪类型
    private long timeStamp = System.currentTimeMillis(); // 当前时间戳
    private String regionId = ""; // Broker所属Region的Id标识
    private String regionName = ""; // Broker所属Region的名称
    private String groupName = "";  // 分组名
    private int costTime = 0; // 耗时
    private boolean isSuccess = true; // 是否成功标识
    private String requestId = MessageClientIDSetter.createUniqID(); // 请求Id
    private int contextCode = 0; // 消费结果返回状态码
    private List<TraceBean> traceBeans; // 轨迹数据集合

public enum TraceType {
    Pub, // 消息发送
    SubBefore, // 消息拉取到客户端，执行业务定义的消费逻辑之前
    SubAfter, // 消费后
}

生产者demo：

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 指定生产者组testMessageTrace，第二个参数enableMsgTrace是启用消息轨迹功能
        DefaultMQProducer producer = new DefaultMQProducer("testMessageTrace", true);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    "VV",
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        producer.shutdown();
    }
}

//DefaultMQProducer.java
public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace) {
    this(null, producerGroup, null, enableMsgTrace, null);
}
public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook,
    boolean enableMsgTrace, final String customizedTraceTopic) {
    this.namespace = namespace;
    this.producerGroup = producerGroup;
    defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    //是否开启消息轨迹
    if (enableMsgTrace) {
        try {
            // 构造异步追踪消息处理分发器
            AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);
            dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
            traceDispatcher = dispatcher;
            // 注入钩子，SendMessageTraceHookImpl实现了消息生产者发送消息的钩子实现
            this.getDefaultMQProducerImpl().registerSendMessageHook(
                new SendMessageTraceHookImpl(traceDispatcher));
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
        }
    }
}
@Override
public void start() throws MQClientException {
    this.setProducerGroup(withNamespace(this.producerGroup));
    // 启动生产者
    this.defaultMQProducerImpl.start();
    // 如果消息追踪分发器不为空，则启动
    if (null != traceDispatcher) {
        try {
            traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
        } catch (MQClientException e) {
            log.warn("trace dispatcher start failed ", e);
        }
    }
}

异步追踪消息处理分发器接口定义：

public interface TraceDispatcher {
    // 初始化异步转换数据模块
    void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException;
    // 追加数据
    boolean append(Object ctx);
    // 数据写入flush
    void flush() throws IOException;
    // 关闭钩子
    void shutdown();
}

异步追踪消息处理分发器接口实现：

public class AsyncTraceDispatcher implements TraceDispatcher {
    private final int queueSize; // 队列大小，默认2048
    private final int batchSize; // 批次大小，默认100
    private final int maxMsgSize; // 最大消息大小，默认128000=128k
    private final DefaultMQProducer traceProducer; // 生产者
    private final ThreadPoolExecutor traceExecutor; // 线程池
    private AtomicLong discardCount; // 最新log丢弃数量
    private Thread worker; // 工作线程
    private ArrayBlockingQueue<TraceContext> traceContextQueue; // 队列：存放消息追踪上下文TraceContext
    private ArrayBlockingQueue<Runnable> appenderQueue; // 追加队列？
    private volatile Thread shutDownHook; // 关闭构造线程
    private volatile boolean stopped = false; // 是否停止
    private DefaultMQProducerImpl hostProducer; // 生产者实现类
    private DefaultMQPushConsumerImpl hostConsumer; // 消费者实现类
    private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex(); // 存放发送到哪个Queue的数据
    private String dispatcherId = UUID.randomUUID().toString(); // 分发处理器ID，随机UUID生成
    private String traceTopicName; // 跟踪主题名称
    private AtomicBoolean isStarted = new AtomicBoolean(false); // 是否启动
    private AccessChannel accessChannel = AccessChannel.LOCAL; // 可以指定区分是Cloud或者Local方式
    // 构造初始化参数
    public AsyncTraceDispatcher(String traceTopicName, RPCHook rpcHook) {
        this.queueSize = 2048;
        this.batchSize = 100;
        this.maxMsgSize = 128000;
        this.discardCount = new AtomicLong(0L);
        this.traceContextQueue = new ArrayBlockingQueue<TraceContext>(1024);
        this.appenderQueue = new ArrayBlockingQueue<Runnable>(queueSize);
        // 如果没有指定消息轨迹主题，那么默认是RMQ_SYS_TRACE_TOPIC
        if (!UtilAll.isBlank(traceTopicName)) {
            this.traceTopicName = traceTopicName;
        } else {
            this.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
        }
        // 构造线程池，最小线程数10，最大20
        this.traceExecutor = new ThreadPoolExecutor(//
            10, //
            20, //
            1000 * 60, //
            TimeUnit.MILLISECONDS, //
            this.appenderQueue, //
            new ThreadFactoryImpl("MQTraceSendThread_"));
        // 构造追踪生产者
        traceProducer = getAndCreateTraceProducer(rpcHook);
    }

    // 异步消息追踪分发器启动，由前面知道生产者启动时会判断分发器不为空就启动当前方法
    public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException {
        // 进行CAS防止并发启动多次，启动traceProducer，将消息发往Broker，也就是相当于正常的生产者发送消息
        if (isStarted.compareAndSet(false, true)) {
            traceProducer.setNamesrvAddr(nameSrvAddr);
            traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
            traceProducer.start();
        }
        this.accessChannel = accessChannel;
        // 启动工作线程，并且设置为守护线程。AsyncRunnable是工作线程主要的逻辑处理实现
        this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
        this.worker.setDaemon(true);
        this.worker.start();
        // 注入ShutDown钩子，分发器关时需要进行相关的清理工作
        this.registerShutDownHook();
    }

    // 构建追踪Producer
    private DefaultMQProducer getAndCreateTraceProducer(RPCHook rpcHook) {
        DefaultMQProducer traceProducerInstance = this.traceProducer;
        if (traceProducerInstance == null) {
            traceProducerInstance = new DefaultMQProducer(rpcHook);
            // 生产者组 _INNER_TRACE_PRODUCER
            traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);
            // 设置发送超时5秒
            traceProducerInstance.setSendMsgTimeout(5000);
            traceProducerInstance.setVipChannelEnabled(false);
            // 最大消息大小是 128K-1k = 118k，不知道RocketMQ的开发者为啥是这样设计的？
            traceProducerInstance.setMaxMessageSize(maxMsgSize - 10 * 1000);
        }
        return traceProducerInstance;
    }

    // 将追踪上下文存放到队列里
    @Override
    public boolean append(final Object ctx) {
        boolean result = traceContextQueue.offer((TraceContext) ctx);
        if (!result) {
            log.info("buffer full" + discardCount.incrementAndGet() + " ,context is " + ctx);
        }
        return result;
    }

    // flush操作，如果队列还有未处理完成的，那么让当前线程sleep 1毫秒，直到处理完毕。
    @Override
    public void flush() throws IOException {
        // The maximum waiting time for refresh,avoid being written all the time, resulting in failure to return.
        long end = System.currentTimeMillis() + 500;
        while (traceContextQueue.size() > 0 || appenderQueue.size() > 0 && System.currentTimeMillis() <= end) {
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                break;
            }
        }
        log.info("------end trace send " + traceContextQueue.size() + "   " + appenderQueue.size());
    }

    // 关闭操作：停止线程池，停止追踪Producer，移除SHUTDOWN钩子
    @Override
    public void shutdown() {
        this.stopped = true;
        this.traceExecutor.shutdown();
        if (isStarted.get()) {
            traceProducer.shutdown();
        }
        this.removeShutdownHook();
    }

    // 注册SHUTDOWN钩子：调用flush方法，主要是等待队列的数据处理完毕。
    public void registerShutDownHook() {
        if (shutDownHook == null) {
            shutDownHook = new Thread(new Runnable() {
                private volatile boolean hasShutdown = false;
                @Override
                public void run() {
                    synchronized (this) {
                        if (!this.hasShutdown) {
                            try {
                                flush();
                            } catch (IOException e) {
                                log.error("system MQTrace hook shutdown failed ,maybe loss some trace data");
                            }
                        }
                    }
                }
            }, "ShutdownHookMQTrace");
            Runtime.getRuntime().addShutdownHook(shutDownHook);
        }
    }

    public void removeShutdownHook() {
        if (shutDownHook != null) {
            Runtime.getRuntime().removeShutdownHook(shutDownHook);
        }
    }

    // 异步线程
    class AsyncRunnable implements Runnable {
        private boolean stopped;
        @Override
        public void run() {
            while (!stopped) {
                // 分批从队列traceContextQueue弹出数据
                List<TraceContext> contexts = new ArrayList<TraceContext>(batchSize);
                for (int i = 0; i < batchSize; i++) {
                    TraceContext context = null;
                    try {
                        //get trace data element from blocking Queue — traceContextQueue
                        context = traceContextQueue.poll(5, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                    }
                    if (context != null) {
                        contexts.add(context);
                    } else {
                        break;
                    }
                }
                if (contexts.size() > 0) {
                    // 将上下文数据分装成AsyncAppenderRequest来处理
                    AsyncAppenderRequest request = new AsyncAppenderRequest(contexts);
                    // 提交到线程池进行处理
                    traceExecutor.submit(request);
                } else if (AsyncTraceDispatcher.this.stopped) {
                    this.stopped = true;
                }
            }
        }
    }

    // 异步发送轨迹消息线程
    class AsyncAppenderRequest implements Runnable {
        List<TraceContext> contextList;

        public AsyncAppenderRequest(final List<TraceContext> contextList) {
            if (contextList != null) {
                this.contextList = contextList;
            } else {
                this.contextList = new ArrayList<TraceContext>(1);
            }
        }

        @Override
        public void run() {
            sendTraceData(contextList);
        }

        public void sendTraceData(List<TraceContext> contextList) {
            Map<String, List<TraceTransferBean>> transBeanMap = new HashMap<String, List<TraceTransferBean>>();
            for (TraceContext context : contextList) {
                if (context.getTraceBeans().isEmpty()) {
                    continue;
                }
                // Topic value corresponding to original message entity content
                String topic = context.getTraceBeans().get(0).getTopic();
                String regionId = context.getRegionId();
                // Use  original message entity's topic as key
                String key = topic;
                // regionId不为空，将其追加到key
                if (!StringUtils.isBlank(regionId)) {
                    key = key + TraceConstants.CONTENT_SPLITOR + regionId;
                }
                List<TraceTransferBean> transBeanList = transBeanMap.get(key);
                if (transBeanList == null) {
                    transBeanList = new ArrayList<TraceTransferBean>();
                    transBeanMap.put(key, transBeanList);
                }
                // 将上下文数据进行转化组装
                TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);
                transBeanList.add(traceData);
            }
            for (Map.Entry<String, List<TraceTransferBean>> entry : transBeanMap.entrySet()) {
                String[] key = entry.getKey().split(String.valueOf(TraceConstants.CONTENT_SPLITOR));
                String dataTopic = entry.getKey();
                String regionId = null;
                if (key.length > 1) {
                    dataTopic = key[0];
                    regionId = key[1];
                }
                flushData(entry.getValue(), dataTopic, regionId);
            }
        }

        /**
         * 批量发数据
         */
        private void flushData(List<TraceTransferBean> transBeanList, String dataTopic, String regionId) {
            if (transBeanList.size() == 0) {
                return;
            }
            // Temporary buffer
            StringBuilder buffer = new StringBuilder(1024);
            int count = 0;
            Set<String> keySet = new HashSet<String>();

            for (TraceTransferBean bean : transBeanList) {
                // Keyset of message trace includes msgId of or original message
                keySet.addAll(bean.getTransKey());
                buffer.append(bean.getTransData());
                count++;
                // Ensure that the size of the package should not exceed the upper limit.
                if (buffer.length() >= traceProducer.getMaxMessageSize()) {
                    sendTraceDataByMQ(keySet, buffer.toString(), dataTopic, regionId);
                    // Clear temporary buffer after finishing
                    buffer.delete(0, buffer.length());
                    keySet.clear();
                    count = 0;
                }
            }
            // 可能还剩下最后一条数据，需要将其发送
            if (count > 0) {
                sendTraceDataByMQ(keySet, buffer.toString(), dataTopic, regionId);
            }
            transBeanList.clear();
        }

        // 发送轨迹数据
        private void sendTraceDataByMQ(Set<String> keySet, final String data, String dataTopic, String regionId) {
            String traceTopic = traceTopicName;
            if (AccessChannel.CLOUD == accessChannel) {
                traceTopic = TraceConstants.TRACE_TOPIC_PREFIX + regionId;
            }
            final Message message = new Message(traceTopic, data.getBytes());
            // Keyset of message trace includes msgId of or original message
            message.setKeys(keySet);
            try {
                Set<String> traceBrokerSet = tryGetMessageQueueBrokerSet(traceProducer.getDefaultMQProducerImpl(), traceTopic);
                // 发送回调，关注发送异常，打印异常日志
                SendCallback callback = new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                    }
                    @Override
                    public void onException(Throwable e) {
                        log.info("send trace data ,the traceData is " + data);
                    }
                };
                if (traceBrokerSet.isEmpty()) {
                    // 没有轨迹Broker，则直接使用默认方式发送
                    traceProducer.send(message, callback, 5000);
                } else {
                    // 根据轨迹Broker将其发送到对应的MessageQueue
                    traceProducer.send(message, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Set<String> brokerSet = (Set<String>) arg;
                            List<MessageQueue> filterMqs = new ArrayList<MessageQueue>();
                            for (MessageQueue queue : mqs) {
                                if (brokerSet.contains(queue.getBrokerName())) {
                                    filterMqs.add(queue);
                                }
                            }
                            int index = sendWhichQueue.getAndIncrement();
                            int pos = Math.abs(index) % filterMqs.size();
                            if (pos < 0) {
                                pos = 0;
                            }
                            return filterMqs.get(pos);
                        }
                    }, traceBrokerSet, callback);
                }

            } catch (Exception e) {
                log.info("send trace data,the traceData is" + data);
            }
        }

        // 获得消息队列Broker集合：获得topic路由信息，并且将Broker集合返回
        private Set<String> tryGetMessageQueueBrokerSet(DefaultMQProducerImpl producer, String topic) {
            Set<String> brokerSet = new HashSet<String>();
            TopicPublishInfo topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
            if (null == topicPublishInfo || !topicPublishInfo.ok()) {
                producer.getTopicPublishInfoTable().putIfAbsent(topic, new TopicPublishInfo());
                producer.getmQClientFactory().updateTopicRouteInfoFromNameServer(topic);
                topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
            }
            if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
                for (MessageQueue queue : topicPublishInfo.getMessageQueueList()) {
                    brokerSet.add(queue.getBrokerName());
                }
            }
            return brokerSet;
        }
    }

}

生产者消息轨迹启动时序图

发送消息钩子接口：

public interface SendMessageHook {
    String hookName();// 钩子名称
    void sendMessageBefore(final SendMessageContext context);// 发送消息之前处理 
    void sendMessageAfter(final SendMessageContext context);// 发送消息之后处理
}

public class SendMessageContext {
    private String producerGroup; // 生产者组
    private Message message; // 消息
    private MessageQueue mq; // 消息队列
    private String brokerAddr; // Broker地址
    private String bornHost; // 发生主机，即客户端
    private CommunicationMode communicationMode; // 通信模式
    private SendResult sendResult; // 发送结果
    private Exception exception; // 异常
    private Object mqTraceContext; // 轨迹上下文
    private Map<String, String> props; // 额外属性
    private DefaultMQProducerImpl producer; // 生产者实现类
    private MessageType msgType = MessageType.Normal_Msg; // 消息类型
    private String namespace; // 命名空间
    ...
}

发送消息钩子实现：

public class SendMessageTraceHookImpl implements SendMessageHook {
    private TraceDispatcher localDispatcher;
    public SendMessageTraceHookImpl(TraceDispatcher localDispatcher) {
        this.localDispatcher = localDispatcher;
    }
    @Override
    public String hookName() {
        return "SendMessageTraceHook";
    }

    @Override
    public void sendMessageBefore(SendMessageContext context) {
        //如果是轨迹消息本身，那么直接返回无需处理
        if (context == null || 
            context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) {
            return;
        }
        //钩子轨迹上下文TraceContext
        TraceContext tuxeContext = new TraceContext();
        tuxeContext.setTraceBeans(new ArrayList<TraceBean>(1));
        context.setMqTraceContext(tuxeContext);
        tuxeContext.setTraceType(TraceType.Pub);
        tuxeContext.setGroupName(NamespaceUtil.withoutNamespace(context.getProducerGroup()));
        //构建轨迹数据实体
        TraceBean traceBean = new TraceBean();
        traceBean.setTopic(NamespaceUtil.withoutNamespace(context.getMessage().getTopic()));
        traceBean.setTags(context.getMessage().getTags());
        traceBean.setKeys(context.getMessage().getKeys());
        traceBean.setStoreHost(context.getBrokerAddr());
        traceBean.setBodyLength(context.getMessage().getBody().length);
        traceBean.setMsgType(context.getMsgType());
        traceBean.setSendTime(System.currentTimeMillis());
        tuxeContext.getTraceBeans().add(traceBean);
    }

    @Override
    public void sendMessageAfter(SendMessageContext context) {
        //如果是轨迹消息本身，那么直接返回无需处理
        if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())
            || context.getMqTraceContext() == null) {
            return;
        }
        if (context.getSendResult() == null) {
            return;
        }
        if (context.getSendResult().getRegionId() == null
            || !context.getSendResult().isTraceOn()) {
            // if switch is false,skip it
            return;
        }
        // 通过Context获取sendMessageBefore生成的轨迹上下文TraceContext
        TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();
        TraceBean traceBean = tuxeContext.getTraceBeans().get(0);
        // 计算耗时
        int costTime = (int) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size());
        tuxeContext.setCostTime(costTime);
        if (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) {
            tuxeContext.setSuccess(true);
        } else {
            tuxeContext.setSuccess(false);
        }
        tuxeContext.setRegionId(context.getSendResult().getRegionId());
        traceBean.setMsgId(context.getSendResult().getMsgId());
        traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());
        // 设置存储时间storeTime，这个数值是估值=时间戳+耗时/2
        traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / 2);
        // 将上下文数据追加到分发器中的队列里
        localDispatcher.append(tuxeContext);
    }

通过上面的钩子接口，那我们知道发生消息钩子有before、after两个方法，进行将消息发送出去之后进行前后的处理逻辑，计算耗时、获得发送结果等操作。

从上面我们大致了解了生产者发送消息钩子的实现机制，但是似乎我们还没看到在哪里调用钩子进行相应的代码逻辑处理，下面继续看源码：

由Producer消息发送流程得知，我们这里只要关注sendKernelImpl这个核心方法即可。

// DefaultMQProducerImpl#sendKernelImpl
private SendResult sendKernelImpl(final Message msg,
                                  final MessageQueue mq,
                                  final CommunicationMode communicationMode,
                                  final SendCallback sendCallback,
                                  final TopicPublishInfo topicPublishInfo,
                                  final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
            ...
            // 判断是否有发送消息钩子,是则进行发送消息钩子Before处理
            if (this.hasSendMessageHook()) {
                context = new SendMessageContext();
                context.setProducer(this);
                context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
                context.setCommunicationMode(communicationMode);
                context.setBornHost(this.defaultMQProducer.getClientIP());
                context.setBrokerAddr(brokerAddr);
                context.setMessage(msg);
                context.setMq(mq);
                context.setNamespace(this.defaultMQProducer.getNamespace());
                String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
                if (isTrans != null && isTrans.equals("true")) {
                    context.setMsgType(MessageType.Trans_Msg_Half);
                }
                if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
                    context.setMsgType(MessageType.Delay_Msg);
                }
                this.executeSendMessageHookBefore(context);
            }
            // 判断是否有发送消息钩子,是则进行发送消息钩子After处理
            if (this.hasSendMessageHook()) {
                context.setSendResult(sendResult);
                this.executeSendMessageHookAfter(context);
            }
      ...            
}

注入生产者发送消息钩子：
//DefaultMQProducerImpl.java
private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
// 将发送消息钩子存放在sendMessageHookList列表
public void registerSendMessageHook(final SendMessageHook hook) {
    this.sendMessageHookList.add(hook);
    log.info("register sendMessage Hook, {}", hook.hookName());
}
public boolean hasSendMessageHook() {
    return !this.sendMessageHookList.isEmpty();
}
public void executeSendMessageHookBefore(final SendMessageContext context) {
    if (!this.sendMessageHookList.isEmpty()) {
        for (SendMessageHook hook : this.sendMessageHookList) {
            try {
                hook.sendMessageBefore(context);
            } catch (Throwable e) {
                log.warn("failed to executeSendMessageHookBefore", e);
            }
        }
    }
}
public void executeSendMessageHookAfter(final SendMessageContext context) {
    if (!this.sendMessageHookList.isEmpty()) {
        for (SendMessageHook hook : this.sendMessageHookList) {
            try {
                hook.sendMessageAfter(context);
            } catch (Throwable e) {
                log.warn("failed to executeSendMessageHookAfter", e);
            }
        }
    }
}

// TopicConfigManager.java
public TopicConfigManager(BrokerController brokerController) {
    ...
    {
        // Broker配置开启消息轨迹，。那么会自动创建消息轨迹topic，读写队列数都是1。
        if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {
            String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
            TopicConfig topicConfig = new TopicConfig(topic);
            this.systemTopicList.add(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
        }
    }
    ...
}

public class ConsumeMessageTraceHookImpl implements ConsumeMessageHook {
    private TraceDispatcher localDispatcher;
    public ConsumeMessageTraceHookImpl(TraceDispatcher localDispatcher) {
        this.localDispatcher = localDispatcher;
    }
    @Override
    public String hookName() {
        return "ConsumeMessageTraceHook";
    }

    // 消息拉在客户端，并且在消费业务消息之前所处理
    @Override
    public void consumeMessageBefore(ConsumeMessageContext context) {
        // 消息列表为空则不处理
        if (context == null || context.getMsgList() == null || context.getMsgList().isEmpty()) {
            return;
        }
        TraceContext traceContext = new TraceContext();
        context.setMqTraceContext(traceContext);
        traceContext.setTraceType(TraceType.SubBefore);// 轨迹类型：SubBefore
        traceContext.setGroupName(NamespaceUtil.withoutNamespace(context.getConsumerGroup()));//
        List<TraceBean> beans = new ArrayList<TraceBean>();
        for (MessageExt msg : context.getMsgList()) {
            if (msg == null) {
                continue;
            }
            String regionId = msg.getProperty(MessageConst.PROPERTY_MSG_REGION);
            String traceOn = msg.getProperty(MessageConst.PROPERTY_TRACE_SWITCH);
            // 从消息属性中获取是否禁用轨迹消息开关，如果设置false，则不处理
            if (traceOn != null && traceOn.equals("false")) {
                continue;
            }
            TraceBean traceBean = new TraceBean();
            traceBean.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic()));//
            traceBean.setMsgId(msg.getMsgId());//
            traceBean.setTags(msg.getTags());//
            traceBean.setKeys(msg.getKeys());//
            traceBean.setStoreTime(msg.getStoreTimestamp());//
            traceBean.setBodyLength(msg.getStoreSize());//
            traceBean.setRetryTimes(msg.getReconsumeTimes());//
            traceContext.setRegionId(regionId);//
            beans.add(traceBean);
        }
        if (beans.size() > 0) {
            traceContext.setTraceBeans(beans);
            traceContext.setTimeStamp(System.currentTimeMillis());
            localDispatcher.append(traceContext);
        }
    }

    // 业务消息处理完之后进行处理
    @Override
    public void consumeMessageAfter(ConsumeMessageContext context) {
        // 消息列表为空则不处理
        if (context == null || context.getMsgList() == null || context.getMsgList().isEmpty()) {
            return;
        }
        TraceContext subBeforeContext = (TraceContext) context.getMqTraceContext();
        if (subBeforeContext.getTraceBeans() == null || subBeforeContext.getTraceBeans().size() < 1) {
            // subbefore为空或其TraceBean列表为空，则不处理
            return;
        }
        TraceContext subAfterContext = new TraceContext();
        subAfterContext.setTraceType(TraceType.SubAfter);// 轨迹类型为：SubAfter
        subAfterContext.setRegionId(subBeforeContext.getRegionId());//
        subAfterContext.setGroupName(NamespaceUtil.withoutNamespace(subBeforeContext.getGroupName()));//
        subAfterContext.setRequestId(subBeforeContext.getRequestId());//
        subAfterContext.setSuccess(context.isSuccess());//

        // 计算耗时：（当前时间-subBefore时间戳）/ 消息列表大小，所以这个值也只是平均值，不是绝对的
        int costTime = (int) ((System.currentTimeMillis() - subBeforeContext.getTimeStamp()) / context.getMsgList().size());
        subAfterContext.setCostTime(costTime);//
        subAfterContext.setTraceBeans(subBeforeContext.getTraceBeans());
        String contextType = context.getProps().get(MixAll.CONSUME_CONTEXT_TYPE);
        if (contextType != null) {
            subAfterContext.setContextCode(ConsumeReturnType.valueOf(contextType).ordinal());
        }
        localDispatcher.append(subAfterContext);
    }
}