#491 增加消息id被谁消费api，扩展id查询命令

vintagewang · vintagewang · commit bf1ad7134820 · 2014-08-12T02:27:11.000+08:00
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java
@@ -50,6 +50,7 @@
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
 import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
+import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
 
 
 /**
@@ -418,4 +419,12 @@ public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup,
             MQBrokerException {
         return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, msgId);
     }
+
+
+    @Override
+    public List<MessageTrack> messageTrackDetail(MessageExt msg) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
 }
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -21,8 +21,10 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
 
@@ -43,6 +45,7 @@
 import com.alibaba.rocketmq.common.admin.OffsetWrapper;
 import com.alibaba.rocketmq.common.admin.RollbackStats;
 import com.alibaba.rocketmq.common.admin.TopicStatsTable;
+import com.alibaba.rocketmq.common.filter.FilterAPI;
 import com.alibaba.rocketmq.common.help.FAQUrl;
 import com.alibaba.rocketmq.common.message.MessageExt;
 import com.alibaba.rocketmq.common.message.MessageQueue;
@@ -59,16 +62,20 @@
 import com.alibaba.rocketmq.common.protocol.body.QueueTimeSpan;
 import com.alibaba.rocketmq.common.protocol.body.TopicList;
 import com.alibaba.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import com.alibaba.rocketmq.common.protocol.route.BrokerData;
 import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
 import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
 import com.alibaba.rocketmq.remoting.RPCHook;
+import com.alibaba.rocketmq.remoting.common.RemotingHelper;
 import com.alibaba.rocketmq.remoting.common.RemotingUtil;
 import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
 import com.alibaba.rocketmq.remoting.exception.RemotingConnectException;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
 import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
+import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
+import com.alibaba.rocketmq.tools.admin.api.TrackType;
 
 
 /**
@@ -727,4 +734,87 @@ public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup,
         return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(
             RemotingUtil.socketAddress2String(msg.getStoreHost()), consumerGroup, clientId, msgId, 10000);
     }
+
+
+    public boolean consumed(final MessageExt msg, final String group) throws RemotingException,
+            MQClientException, InterruptedException, MQBrokerException {
+        // 查询消费进度
+        ConsumeStats cstats = this.examineConsumeStats(group);
+
+        ClusterInfo ci = this.examineBrokerClusterInfo();
+
+        Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
+        while (it.hasNext()) {
+            Entry<MessageQueue, OffsetWrapper> next = it.next();
+            MessageQueue mq = next.getKey();
+            if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
+                BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
+                String addr = brokerData.getBrokerAddrs().get(0);
+                if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
+                    if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
+                        return true;
+                    }
+                }
+            }
+        }
+
+        return false;
+    }
+
+
+    @Override
+    public List<MessageTrack> messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException,
+            InterruptedException, MQBrokerException {
+        List<MessageTrack> result = new ArrayList<MessageTrack>();
+
+        GroupList groupList = this.queryTopicConsumeByWho(msg.getTopic());
+
+        for (String group : groupList.getGroupList()) {
+
+            // 查询连接
+            MessageTrack mt = new MessageTrack();
+            mt.setConsumerGroup(group);
+            mt.setTrackType(TrackType.UNKNOW_EXCEPTION);
+            try {
+                ConsumerConnection cc = this.examineConsumerConnectionInfo(group);
+                switch (cc.getConsumeType()) {
+                case CONSUME_ACTIVELY:
+                    mt.setTrackType(TrackType.SUBSCRIBED_BUT_PULL);
+                    break;
+                case CONSUME_PASSIVELY:
+                    boolean ifConsumed = this.consumed(msg, group);
+                    if (ifConsumed) {
+                        mt.setTrackType(TrackType.SUBSCRIBED_AND_CONSUMED);
+
+                        // 查看订阅关系是否匹配
+                        Iterator<Entry<String, SubscriptionData>> it =
+                                cc.getSubscriptionTable().entrySet().iterator();
+                        while (it.hasNext()) {
+                            Entry<String, SubscriptionData> next = it.next();
+                            if (next.getKey().equals(msg.getTopic())) {
+                                if (next.getValue().getTagsSet().contains(msg.getTags()) //
+                                        || next.getValue().getTagsSet().contains("*")) {
+
+                                }
+                                else {
+                                    mt.setTrackType(TrackType.SUBSCRIBED_BUT_FILTERD);
+                                }
+                            }
+                        }
+                    }
+                    else {
+                        mt.setTrackType(TrackType.SUBSCRIBED_AND_NOT_CONSUME);
+                    }
+                    break;
+                default:
+                    break;
+                }
+            }
+            catch (Exception e) {
+                mt.setExceptionDesc(RemotingHelper.exceptionSimpleDesc(e));
+            }
+        }
+
+        return null;
+    }
 }
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/MQAdminExt.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/MQAdminExt.java
@@ -28,6 +28,7 @@
 import com.alibaba.rocketmq.common.admin.ConsumeStats;
 import com.alibaba.rocketmq.common.admin.RollbackStats;
 import com.alibaba.rocketmq.common.admin.TopicStatsTable;
+import com.alibaba.rocketmq.common.message.MessageExt;
 import com.alibaba.rocketmq.common.message.MessageQueue;
 import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
 import com.alibaba.rocketmq.common.protocol.body.ConsumeByWho;
@@ -46,6 +47,7 @@
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
 import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
+import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
 
 
 /**
@@ -586,4 +588,18 @@ public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup,
             String clientId, //
             String msgId) throws RemotingException, MQClientException, InterruptedException,
             MQBrokerException;
+
+
+    /**
+     * 查询消息被谁消费了
+     * 
+     * @param msg
+     * @return
+     * @throws RemotingException
+     * @throws MQClientException
+     * @throws InterruptedException
+     * @throws MQBrokerException
+     */
+    public List<MessageTrack> messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException,
+            InterruptedException, MQBrokerException;
 }
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/api/MessageTrack.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/api/MessageTrack.java
@@ -0,0 +1,46 @@
+package com.alibaba.rocketmq.tools.admin.api;
+
+
+
+public class MessageTrack {
+    private String consumerGroup;
+    private TrackType trackType;
+    private String exceptionDesc;
+
+
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+
+    public void setConsumerGroup(String consumerGroup) {
+        this.consumerGroup = consumerGroup;
+    }
+
+
+    public TrackType getTrackType() {
+        return trackType;
+    }
+
+
+    public void setTrackType(TrackType trackType) {
+        this.trackType = trackType;
+    }
+
+
+    public String getExceptionDesc() {
+        return exceptionDesc;
+    }
+
+
+    public void setExceptionDesc(String exceptionDesc) {
+        this.exceptionDesc = exceptionDesc;
+    }
+
+
+    @Override
+    public String toString() {
+        return "MessageTrack [consumerGroup=" + consumerGroup + ", trackType=" + trackType
+                + ", exceptionDesc=" + exceptionDesc + "]";
+    }
+}
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/api/TrackType.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/api/TrackType.java
@@ -0,0 +1,14 @@
+package com.alibaba.rocketmq.tools.admin.api;
+
+public enum TrackType {
+    // 订阅了，而且消费了（Offset越过了）
+    SUBSCRIBED_AND_CONSUMED,
+    // 订阅了，但是被过滤掉了
+    SUBSCRIBED_BUT_FILTERD,
+    // 订阅了，但是是PULL，结果未知
+    SUBSCRIBED_BUT_PULL,
+    // 订阅了，但是没有消费（Offset小）
+    SUBSCRIBED_AND_NOT_CONSUME,
+    // 未知异常
+    UNKNOW_EXCEPTION,
+}
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java
@@ -19,6 +19,7 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.util.List;
 
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
@@ -33,6 +34,7 @@
 import com.alibaba.rocketmq.remoting.common.RemotingHelper;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
+import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
 import com.alibaba.rocketmq.tools.command.SubCommand;
 
 
@@ -151,6 +153,11 @@ public static void queryById(final DefaultMQAdminExt admin, final String msgId)
             "Message Body Path:",//
             bodyTmpFilePath//
             );
+
+        List<MessageTrack> mtdList = admin.messageTrackDetail(msg);
+        for (MessageTrack mt : mtdList) {
+            System.out.println(mt);
+        }
     }
 
 
