添加服务端消息轨迹

Author: YangJodie

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/BrokerController.java

@@ -15,46 +15,22 @@
  */
 package com.alibaba.rocketmq.broker;
 
-import java.io.IOException;
-import java.util.Properties;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.rocketmq.broker.client.ClientHousekeepingService;
-import com.alibaba.rocketmq.broker.client.ConsumerIdsChangeListener;
-import com.alibaba.rocketmq.broker.client.ConsumerManager;
-import com.alibaba.rocketmq.broker.client.DefaultConsumerIdsChangeListener;
-import com.alibaba.rocketmq.broker.client.ProducerManager;
+import com.alibaba.rocketmq.broker.client.*;
 import com.alibaba.rocketmq.broker.client.net.Broker2Client;
 import com.alibaba.rocketmq.broker.client.rebalance.RebalanceLockManager;
 import com.alibaba.rocketmq.broker.filtersrv.FilterServerManager;
 import com.alibaba.rocketmq.broker.longpolling.PullRequestHoldService;
+import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageHook;
+import com.alibaba.rocketmq.broker.mqtrace.SendMessageHook;
 import com.alibaba.rocketmq.broker.offset.ConsumerOffsetManager;
 import com.alibaba.rocketmq.broker.out.BrokerOuterAPI;
-import com.alibaba.rocketmq.broker.processor.AdminBrokerProcessor;
-import com.alibaba.rocketmq.broker.processor.ClientManageProcessor;
-import com.alibaba.rocketmq.broker.processor.EndTransactionProcessor;
-import com.alibaba.rocketmq.broker.processor.PullMessageProcessor;
-import com.alibaba.rocketmq.broker.processor.QueryMessageProcessor;
-import com.alibaba.rocketmq.broker.processor.SendMessageProcessor;
+import com.alibaba.rocketmq.broker.processor.*;
 import com.alibaba.rocketmq.broker.slave.SlaveSynchronize;
 import com.alibaba.rocketmq.broker.stats.BrokerStats;
 import com.alibaba.rocketmq.broker.stats.BrokerStatsManager;
 import com.alibaba.rocketmq.broker.subscription.SubscriptionGroupManager;
 import com.alibaba.rocketmq.broker.topic.TopicConfigManager;
-import com.alibaba.rocketmq.common.BrokerConfig;
-import com.alibaba.rocketmq.common.DataVersion;
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.ThreadFactoryImpl;
-import com.alibaba.rocketmq.common.UtilAll;
+import com.alibaba.rocketmq.common.*;
 import com.alibaba.rocketmq.common.constant.LoggerName;
 import com.alibaba.rocketmq.common.namesrv.RegisterBrokerResult;
 import com.alibaba.rocketmq.common.protocol.RequestCode;
@@ -69,6 +45,14 @@
 import com.alibaba.rocketmq.store.MessageStore;
 import com.alibaba.rocketmq.store.config.BrokerRole;
 import com.alibaba.rocketmq.store.config.MessageStoreConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.*;
 
 
 /**
@@ -346,7 +330,8 @@ public void registerProcessor() {
         /**
          * SendMessageProcessor
          */
-        NettyRequestProcessor sendProcessor = new SendMessageProcessor(this);
+        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
+        sendProcessor.registerSendMessageHook(sendMessageHookList);
         this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor,
             this.sendMessageExecutor);
         this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor,
@@ -357,6 +342,7 @@ public void registerProcessor() {
          */
         this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor,
             this.pullMessageExecutor);
+        this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
 
         /**
          * QueryMessageProcessor
@@ -387,8 +373,9 @@ public void registerProcessor() {
         /**
          * Default
          */
-        this.remotingServer
-            .registerDefaultProcessor(new AdminBrokerProcessor(this), this.adminBrokerExecutor);
+        AdminBrokerProcessor adminProcessor = new AdminBrokerProcessor(this);
+        adminProcessor.registerConsumeMessageHook(this.consumeMessageHookList);
+        this.remotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);
     }
 
 
@@ -765,6 +752,24 @@ public void run() {
         }, 5, TimeUnit.MINUTES);
     }
 
+    // 注册发送消息轨迹 hook
+    private final List<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
+
+
+    public void registerSendMessageHook(final SendMessageHook hook) {
+        this.sendMessageHookList.add(hook);
+        log.info("register SendMessageHook Hook, {}", hook.hookName());
+    }
+
+    // 注册消费消息轨迹 hook
+    private final List<ConsumeMessageHook> consumeMessageHookList = new ArrayList<ConsumeMessageHook>();
+
+
+    public void registerConsumeMessageHook(final ConsumeMessageHook hook) {
+        this.consumeMessageHookList.add(hook);
+        log.info("register ConsumeMessageHook Hook, {}", hook.hookName());
+    }
+
 
     public void registerServerRPCHook(RPCHook rpcHook) {
         getRemotingServer().registerRPCHook(rpcHook);

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/mqtrace/ConsumeMessageContext.java

@@ -0,0 +1,110 @@
+/**
+ * Copyright (C) 2010-2013 Alibaba Group Holding Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.rocketmq.broker.mqtrace;
+
+import java.util.Map;
+
+
+public class ConsumeMessageContext {
+    private String consumerGroup;
+    private String topic;
+    private Integer queueId;
+    private String clientHost;
+    private String storeHost;
+    private Map<Long, String> messageIds;
+    private boolean success;
+    private String status;
+
+
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+
+    public void setConsumerGroup(String consumerGroup) {
+        this.consumerGroup = consumerGroup;
+    }
+
+
+    public String getTopic() {
+        return topic;
+    }
+
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+
+    public Integer getQueueId() {
+        return queueId;
+    }
+
+
+    public void setQueueId(Integer queueId) {
+        this.queueId = queueId;
+    }
+
+
+    public String getClientHost() {
+        return clientHost;
+    }
+
+
+    public void setClientHost(String clientHost) {
+        this.clientHost = clientHost;
+    }
+
+
+    public Map<Long, String> getMessageIds() {
+        return messageIds;
+    }
+
+
+    public void setMessageIds(Map<Long, String> messageIds) {
+        this.messageIds = messageIds;
+    }
+
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+
+    public void setSuccess(boolean success) {
+        this.success = success;
+    }
+
+
+    public String getStatus() {
+        return status;
+    }
+
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+
+
+    public String getStoreHost() {
+        return storeHost;
+    }
+
+
+    public void setStoreHost(String storeHost) {
+        this.storeHost = storeHost;
+    }
+}

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/mqtrace/ConsumeMessageHook.java

@@ -0,0 +1,28 @@
+/**
+ * Copyright (C) 2010-2013 Alibaba Group Holding Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.rocketmq.broker.mqtrace;
+
+
+
+public interface ConsumeMessageHook {
+    public String hookName();
+
+
+    public void consumeMessageBefore(final ConsumeMessageContext context);
+
+
+    public void consumeMessageAfter(final ConsumeMessageContext context);
+}