发送消息网络包优化

Author: vintagewang

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

@@ -334,6 +334,8 @@ public void registerProcessor() {
         sendProcessor.registerSendMessageHook(sendMessageHookList);
         this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor,
             this.sendMessageExecutor);
+        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor,
+            this.sendMessageExecutor);
         this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor,
             this.sendMessageExecutor);
 

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/processor/SendMessageProcessor.java

@@ -15,6 +15,16 @@
  */
 package com.alibaba.rocketmq.broker.processor;
 
+import io.netty.channel.ChannelHandlerContext;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.List;
+import java.util.Random;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.alibaba.rocketmq.broker.BrokerController;
 import com.alibaba.rocketmq.broker.mqtrace.SendMessageContext;
 import com.alibaba.rocketmq.broker.mqtrace.SendMessageHook;
@@ -33,6 +43,7 @@
 import com.alibaba.rocketmq.common.protocol.ResponseCode;
 import com.alibaba.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
 import com.alibaba.rocketmq.common.protocol.header.SendMessageRequestHeader;
+import com.alibaba.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
 import com.alibaba.rocketmq.common.protocol.header.SendMessageResponseHeader;
 import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
 import com.alibaba.rocketmq.common.sysflag.MessageSysFlag;
@@ -43,14 +54,6 @@
 import com.alibaba.rocketmq.store.MessageExtBrokerInner;
 import com.alibaba.rocketmq.store.PutMessageResult;
 import com.alibaba.rocketmq.store.config.StorePathConfigHelper;
-import io.netty.channel.ChannelHandlerContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.util.List;
-import java.util.Random;
 
 
 /**
@@ -79,14 +82,27 @@ public SendMessageProcessor(final BrokerController brokerController) {
     @Override
     public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
             throws RemotingCommandException {
+        SendMessageRequestHeaderV2 requestHeaderV2 = null;
+
         switch (request.getCode()) {
+        case RequestCode.SEND_MESSAGE_V2:
+            requestHeaderV2 =
+                    (SendMessageRequestHeaderV2) request
+                        .decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
         case RequestCode.SEND_MESSAGE:
             SendMessageContext context = null;
-            // 消息轨迹：记录到达 broker 的消息
-            if (this.hasSendMessageHook()) {
-                final SendMessageRequestHeader requestHeader =
+            SendMessageRequestHeader requestHeader = null;
+
+            if (null == requestHeaderV2) {
+                requestHeader =
                         (SendMessageRequestHeader) request
                             .decodeCommandCustomHeader(SendMessageRequestHeader.class);
+            }
+            else {
+                requestHeader = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV1(requestHeaderV2);
+            }
+            // 消息轨迹：记录到达 broker 的消息
+            if (this.hasSendMessageHook()) {
                 context = new SendMessageContext();
                 context.setProducerGroup(requestHeader.getProducerGroup());
                 context.setTopic(requestHeader.getTopic());
@@ -96,7 +112,7 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand
                 this.executeSendMessageHookBefore(context);
             }
 
-            final RemotingCommand response = this.sendMessage(ctx, request);
+            final RemotingCommand response = this.sendMessage(ctx, request, requestHeader);
 
             // 消息轨迹：记录发送成功的消息
             if (this.hasSendMessageHook()) {
@@ -162,7 +178,7 @@ private RemotingCommand consumerSendMsgBack(final ChannelHandlerContext ctx, fin
         // 检查topic权限
         if (!PermName.isWriteable(topicConfig.getPerm())) {
             response.setCode(ResponseCode.NO_PERMISSION);
-            response.setRemark("the topic[" + newTopic + "] sending message is forbidden");
+            response.setRemark(String.format("the topic[%s] sending message is forbidden", newTopic));
             return response;
         }
 
@@ -278,14 +294,12 @@ private String diskUtil() {
     }
 
 
-    private RemotingCommand sendMessage(final ChannelHandlerContext ctx, final RemotingCommand request)
-            throws RemotingCommandException {
+    private RemotingCommand sendMessage(final ChannelHandlerContext ctx, final RemotingCommand request,
+            final SendMessageRequestHeader requestHeader) throws RemotingCommandException {
         final RemotingCommand response =
                 RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
         final SendMessageResponseHeader responseHeader =
                 (SendMessageResponseHeader) response.readCustomHeader();
-        final SendMessageRequestHeader requestHeader =
-                (SendMessageRequestHeader) request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
 
         // 由于有直接返回的逻辑，所以必须要设置
         response.setOpaque(request.getOpaque());

rocketmq-common/src/main/java/com/alibaba/rocketmq/common/protocol/RequestCode.java

@@ -151,4 +151,7 @@ public class RequestCode {
     // 通过Broker直接向某个Consumer发送一条消息，并立刻消费，返回结果给broker，再返回给调用方
     // 2014-08-11 Add By shijia
     public static final int CONSUME_MESSAGE_DIRECTLY = 308;
+
+    // Broker 发送消息，优化网络数据包
+    public static final int SEND_MESSAGE_V2 = 309;
 }

rocketmq-common/src/main/java/com/alibaba/rocketmq/common/protocol/header/SendMessageRequestHeaderV2.java

@@ -0,0 +1,188 @@
+package com.alibaba.rocketmq.common.protocol.header;
+
+import com.alibaba.rocketmq.remoting.CommandCustomHeader;
+import com.alibaba.rocketmq.remoting.annotation.CFNotNull;
+import com.alibaba.rocketmq.remoting.annotation.CFNullable;
+import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * 为减少网络传输数量准备
+ * 
+ * @author shijia.wxr<[email protected]>
+ */
+public class SendMessageRequestHeaderV2 implements CommandCustomHeader {
+    @CFNotNull
+    private String a;// producerGroup;
+    @CFNotNull
+    private String b;// topic;
+    @CFNotNull
+    private String c;// defaultTopic;
+    @CFNotNull
+    private Integer d;// defaultTopicQueueNums;
+    @CFNotNull
+    private Integer e;// queueId;
+    @CFNotNull
+    private Integer f;// sysFlag;
+    @CFNotNull
+    private Long g;// bornTimestamp;
+    @CFNotNull
+    private Integer h;// flag;
+    @CFNullable
+    private String i;// properties;
+    @CFNullable
+    private Integer j;// reconsumeTimes;
+    @CFNullable
+    private boolean k;// unitMode = false;
+
+
+    @Override
+    public void checkFields() throws RemotingCommandException {
+    }
+
+
+    public static SendMessageRequestHeader createSendMessageRequestHeaderV1(
+            final SendMessageRequestHeaderV2 v2) {
+        SendMessageRequestHeader v1 = new SendMessageRequestHeader();
+        v1.setProducerGroup(v2.a);
+        v1.setTopic(v2.b);
+        v1.setDefaultTopic(v2.c);
+        v1.setDefaultTopicQueueNums(v2.d);
+        v1.setQueueId(v2.e);
+        v1.setSysFlag(v2.f);
+        v1.setBornTimestamp(v2.g);
+        v1.setFlag(v2.h);
+        v1.setProperties(v2.i);
+        v1.setReconsumeTimes(v2.j);
+        v1.setUnitMode(v2.k);
+        return v1;
+    }
+
+
+    public static SendMessageRequestHeaderV2 createSendMessageRequestHeaderV2(
+            final SendMessageRequestHeader v1) {
+        SendMessageRequestHeaderV2 v2 = new SendMessageRequestHeaderV2();
+        v2.a = v1.getProducerGroup();
+        v2.b = v1.getTopic();
+        v2.c = v1.getDefaultTopic();
+        v2.d = v1.getDefaultTopicQueueNums();
+        v2.e = v1.getQueueId();
+        v2.f = v1.getSysFlag();
+        v2.g = v1.getBornTimestamp();
+        v2.h = v1.getFlag();
+        v2.i = v1.getProperties();
+        v2.j = v1.getReconsumeTimes();
+        v2.k = v1.isUnitMode();
+        return v2;
+    }
+
+
+    public String getA() {
+        return a;
+    }
+
+
+    public void setA(String a) {
+        this.a = a;
+    }
+
+
+    public String getB() {
+        return b;
+    }
+
+
+    public void setB(String b) {
+        this.b = b;
+    }
+
+
+    public String getC() {
+        return c;
+    }
+
+
+    public void setC(String c) {
+        this.c = c;
+    }
+
+
+    public Integer getD() {
+        return d;
+    }
+
+
+    public void setD(Integer d) {
+        this.d = d;
+    }
+
+
+    public Integer getE() {
+        return e;
+    }
+
+
+    public void setE(Integer e) {
+        this.e = e;
+    }
+
+
+    public Integer getF() {
+        return f;
+    }
+
+
+    public void setF(Integer f) {
+        this.f = f;
+    }
+
+
+    public Long getG() {
+        return g;
+    }
+
+
+    public void setG(Long g) {
+        this.g = g;
+    }
+
+
+    public Integer getH() {
+        return h;
+    }
+
+
+    public void setH(Integer h) {
+        this.h = h;
+    }
+
+
+    public String getI() {
+        return i;
+    }
+
+
+    public void setI(String i) {
+        this.i = i;
+    }
+
+
+    public Integer getJ() {
+        return j;
+    }
+
+
+    public void setJ(Integer j) {
+        this.j = j;
+    }
+
+
+    public boolean isK() {
+        return k;
+    }
+
+
+    public void setK(boolean k) {
+        this.k = k;
+    }
+}