#490 向指定Consumer发起消费消息请求，运维需要。客户端开发完成。

vintagewang · vintagewang · commit dea1df4a729c · 2014-08-11T21:02:48.000+08:00
diff --git a/rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/MQClientAPIImpl.java b/rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/MQClientAPIImpl.java
@@ -56,6 +56,7 @@
 import com.alibaba.rocketmq.common.protocol.RequestCode;
 import com.alibaba.rocketmq.common.protocol.ResponseCode;
 import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
+import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
 import com.alibaba.rocketmq.common.protocol.body.GetConsumerStatusBody;
@@ -69,6 +70,7 @@
 import com.alibaba.rocketmq.common.protocol.body.ResetOffsetBody;
 import com.alibaba.rocketmq.common.protocol.body.TopicList;
 import com.alibaba.rocketmq.common.protocol.body.UnlockBatchRequestBody;
+import com.alibaba.rocketmq.common.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
 import com.alibaba.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
 import com.alibaba.rocketmq.common.protocol.header.CreateTopicRequestHeader;
 import com.alibaba.rocketmq.common.protocol.header.DeleteSubscriptionGroupRequestHeader;
@@ -2152,6 +2154,41 @@ public ConsumerRunningInfo getConsumerRunningInfo(final String addr, String cons
     }
 
 
+    /**
+     * 通过调用Broker，向指定Consumer发送某条消息，并返回消费结果
+     */
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(final String addr, //
+            String consumerGroup, //
+            String clientId, //
+            String msgId, //
+            final long timeoutMillis) throws RemotingException, MQClientException, InterruptedException {
+        ConsumeMessageDirectlyResultRequestHeader requestHeader =
+                new ConsumeMessageDirectlyResultRequestHeader();
+        requestHeader.setConsumerGroup(consumerGroup);
+        requestHeader.setClientId(clientId);
+
+        RemotingCommand request =
+                RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, requestHeader);
+
+        RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
+        assert response != null;
+        switch (response.getCode()) {
+        case ResponseCode.SUCCESS: {
+            byte[] body = response.getBody();
+            if (body != null) {
+                ConsumeMessageDirectlyResult info =
+                        ConsumeMessageDirectlyResult.decode(body, ConsumeMessageDirectlyResult.class);
+                return info;
+            }
+        }
+        default:
+            break;
+        }
+
+        throw new MQClientException(response.getCode(), response.getRemark());
+    }
+
+
     public String getProjectGroupPrefix() {
         return projectGroupPrefix;
     }
diff --git a/rocketmq-common/src/main/java/com/alibaba/rocketmq/common/protocol/body/ConsumeMessageDirectlyResult.java b/rocketmq-common/src/main/java/com/alibaba/rocketmq/common/protocol/body/ConsumeMessageDirectlyResult.java
@@ -1,6 +1,9 @@
 package com.alibaba.rocketmq.common.protocol.body;
 
-public class ConsumeMessageDirectlyResult {
+import com.alibaba.rocketmq.remoting.protocol.RemotingSerializable;
+
+
+public class ConsumeMessageDirectlyResult extends RemotingSerializable {
     private boolean order = false;
     private boolean autoCommit = true;
     private CMResult consumeResult;
@@ -56,4 +59,12 @@ public long getSpentTimeMills() {
     public void setSpentTimeMills(long spentTimeMills) {
         this.spentTimeMills = spentTimeMills;
     }
+
+
+    @Override
+    public String toString() {
+        return "ConsumeMessageDirectlyResult [order=" + order + ", autoCommit=" + autoCommit
+                + ", consumeResult=" + consumeResult + ", remark=" + remark + ", spentTimeMills="
+                + spentTimeMills + "]";
+    }
 }
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
@@ -34,6 +34,7 @@
 import com.alibaba.rocketmq.common.message.MessageQueue;
 import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
 import com.alibaba.rocketmq.common.protocol.body.ConsumeByWho;
+import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
 import com.alibaba.rocketmq.common.protocol.body.GroupList;
@@ -409,4 +410,12 @@ public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String c
             throws RemotingException, MQClientException, InterruptedException {
         return defaultMQAdminExtImpl.getConsumerRunningInfo(consumerGroup, clientId);
     }
+
+
+    @Override
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, String clientId,
+            String msgId) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, msgId);
+    }
 }
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
@@ -50,6 +50,7 @@
 import com.alibaba.rocketmq.common.protocol.ResponseCode;
 import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
 import com.alibaba.rocketmq.common.protocol.body.ConsumeByWho;
+import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
 import com.alibaba.rocketmq.common.protocol.body.GroupList;
@@ -62,6 +63,7 @@
 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.RemotingUtil;
 import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
 import com.alibaba.rocketmq.remoting.exception.RemotingConnectException;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
@@ -714,4 +716,15 @@ public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String c
         }
         return null;
     }
+
+
+    @Override
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, String clientId,
+            String msgId) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        MessageExt msg = this.viewMessage(msgId);
+
+        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(
+            RemotingUtil.socketAddress2String(msg.getStoreHost()), consumerGroup, clientId, msgId, 10000);
+    }
 }
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
@@ -31,6 +31,7 @@
 import com.alibaba.rocketmq.common.message.MessageQueue;
 import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
 import com.alibaba.rocketmq.common.protocol.body.ConsumeByWho;
+import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
 import com.alibaba.rocketmq.common.protocol.body.GroupList;
@@ -567,4 +568,22 @@ public boolean cleanExpiredConsumerQueueByAddr(String addr) throws RemotingConne
      */
     public ConsumerRunningInfo getConsumerRunningInfo(final String consumerGroup, final String clientId)
             throws RemotingException, MQClientException, InterruptedException;
+
+
+    /**
+     * 向指定Consumer发送某条消息
+     * 
+     * @param consumerGroup
+     * @param clientId
+     * @param msgId
+     * @return
+     * @throws InterruptedException
+     * @throws MQClientException
+     * @throws RemotingException
+     * @throws MQBrokerException
+     */
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, //
+            String clientId, //
+            String msgId) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException;
 }
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
@@ -28,6 +28,7 @@
 import com.alibaba.rocketmq.client.exception.MQClientException;
 import com.alibaba.rocketmq.common.UtilAll;
 import com.alibaba.rocketmq.common.message.MessageExt;
+import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
 import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.remoting.common.RemotingHelper;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
@@ -61,6 +62,14 @@ public Options buildCommandlineOptions(Options options) {
         opt.setRequired(true);
         options.addOption(opt);
 
+        opt = new Option("g", "consumerGroup", true, "consumer group name");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("d", "clientId", true, "The consumer's client id");
+        opt.setRequired(false);
+        options.addOption(opt);
+
         return options;
     }
 
@@ -153,7 +162,16 @@ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
 
         try {
             final String msgId = commandLine.getOptionValue('i').trim();
-            queryById(defaultMQAdminExt, msgId);
+            if (commandLine.hasOption('g') && commandLine.hasOption('d')) {
+                final String consumerGroup = commandLine.getOptionValue('g').trim();
+                final String clientId = commandLine.getOptionValue('d').trim();
+                ConsumeMessageDirectlyResult result =
+                        defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, msgId);
+                System.out.println(result);
+            }
+            else {
+                queryById(defaultMQAdminExt, msgId);
+            }
         }
         catch (Exception e) {
             e.printStackTrace();
