insert-app/src/main/java/db/netty/NettyServer.java

@@ -70,7 +70,8 @@ public class NettyServer {
         group = useMultiThread ? new EpollEventLoopGroup() : new NioEventLoopGroup();
 
         log.info("开始启动 Netty Server，host:{} port: {}", host, port);
-        log.info("Netty-Server isJustReceivePacket:{}", isJustReceivePacket);
+        log.info("***Netty-Server isJustReceivePacket:{}", isJustReceivePacket);
+        log.info("***Netty-Server recvBufferSize:{}", recvBufferSize);
         Bootstrap bootstrap = new Bootstrap();
 
         Encoder encoder = new Encoder();

protocol/src/main/java/db/netty/Decoder.java

@@ -21,9 +21,11 @@ public class Decoder extends MessageToMessageDecoder<DatagramPacket> {
 
     @Override
     protected void decode(ChannelHandlerContext ctx, DatagramPacket datagramPacket, List<Object> out) throws Exception {
-        DecodeStatistic.getInstance().increaseNeedDecodePacketNum(1);
-        if(!decode) return;
+        if(!decode) decodeDebug(ctx, datagramPacket, out);
 
+        decodeReal(ctx, datagramPacket, out);
+    }
+    protected void decodeReal(ChannelHandlerContext ctx, DatagramPacket datagramPacket, List<Object> out) throws Exception {
         ByteBuf udpPacketBytes = datagramPacket.content();
         byte[] bytesReady = new byte[udpPacketBytes.readableBytes()];
         udpPacketBytes.getBytes(0, bytesReady);
@@ -142,6 +144,7 @@ public class Decoder extends MessageToMessageDecoder<DatagramPacket> {
             basePackage = new ListenResponsePackage();
         } else if (operationType == Globals.TYPE_RESPONSE && operationCode == 0x0006) {
             log.debug("解码器转换为监听数据报文");
+            DecodeStatistic.getInstance().increaseNeedDecodePacketNum(1);
             basePackage = new DataResponsePackage();
         } else if (operationType == Globals.TYPE_REQUEST && operationCode == Globals.OPERATION_WRITE &&
                 packageCode == Globals.PACKAGE_LOCK) {
@@ -187,6 +190,29 @@ public class Decoder extends MessageToMessageDecoder<DatagramPacket> {
         }
     }
 
+    protected void decodeDebug(ChannelHandlerContext ctx, DatagramPacket datagramPacket,
+                               List<Object> out) throws Exception {
+        ByteBuf udpPacketBytes = datagramPacket.content();
+        byte[] bytesReady = new byte[udpPacketBytes.readableBytes()];
+        udpPacketBytes.getBytes(0, bytesReady);
+
+        int magic = udpPacketBytes.getInt(0);
+
+        if (magic != Globals.PROBE_MAGIC) {
+            DecodeStatistic.getInstance().increaseNeedDecodePacketNum(1);
+            log.warn("魔数错误，消息为：{}", ConvertUtils.bytes2HexArray(bytesReady));
+            return;
+        }
+
+        short operationType = udpPacketBytes.getShort(8);
+        short operationCode = udpPacketBytes.getShort(12);
+
+        if (operationType == Globals.TYPE_RESPONSE && operationCode == 0x0006) {
+            log.debug("解码器转换为监听数据报文");
+            DecodeStatistic.getInstance().increaseNeedDecodePacketNum(1);
+        }
+    }
+
     public void setShowRecvBytes(boolean showRecvBytes) {
         this.showRecvBytes = showRecvBytes;
     }

protocol/src/main/java/db/tools/DecodeStatistic.java

@@ -10,7 +10,7 @@ import java.util.concurrent.atomic.AtomicLong;
  */
 public class DecodeStatistic {
     private static DecodeStatistic instance = new DecodeStatistic();
-    private AtomicLong needDecodePacketNum = new AtomicLong();
+    private AtomicLong needDecodePacketNum = new AtomicLong(0);
     public static DecodeStatistic getInstance() {
         return instance;
     }