【fix】 自定义netty 接收数据缓冲期控制接收数据大小

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/NettyByteBuf.java

@@ -0,0 +1,18 @@
+package org.framework.lazy.cloud.network.heartbeat.common;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@NoArgsConstructor
+@Setter
+@Getter
+public class NettyByteBuf {
+    // body 长度  data 4
+    public static final int bodyLength = 4;
+    /**
+     * 消息传输数据
+     * byte[] 长度 4
+     */
+    private byte[] data;
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/allocator/NettyRecvByteBufAllocator.java

@@ -0,0 +1,75 @@
+package org.framework.lazy.cloud.network.heartbeat.common.allocator;
+
+import io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator;
+import io.netty.channel.FixedRecvByteBufAllocator;
+import io.netty.channel.RecvByteBufAllocator;
+import lombok.extern.slf4j.Slf4j;
+
+import static io.netty.util.internal.ObjectUtil.checkPositive;
+
+/**
+ * The {@link RecvByteBufAllocator} that always yields the same buffer
+ * size prediction.  This predictor ignores the feed back from the I/O thread.
+ * 接收数据缓冲区
+ *
+ * @see FixedRecvByteBufAllocator
+ */
+@Slf4j
+public class NettyRecvByteBufAllocator extends DefaultMaxMessagesRecvByteBufAllocator {
+
+    private final int bufferSize;
+
+    private final class HandleImpl extends MaxMessageHandle {
+        private final int bufferSize;
+
+        HandleImpl(int bufferSize) {
+            this.bufferSize = bufferSize;
+        }
+
+        @Override
+        public int guess() {
+//            log.info("guess :{}", bufferSize);
+            return bufferSize;
+        }
+
+        @Override
+        public void attemptedBytesRead(int bytes) {
+//            log.info("attemptedBytesRead:{}", bytes);
+            super.attemptedBytesRead(bytes);
+        }
+
+        @Override
+        public int attemptedBytesRead() {
+            int attemptedBytesRead = super.attemptedBytesRead();
+//            log.info("attemptedBytesRead result:{}", attemptedBytesRead);
+            return attemptedBytesRead;
+        }
+
+        @Override
+        public void lastBytesRead(int bytes) {
+            super.lastBytesRead(bytes);
+//            log.info("lastBytesRead result:{}", bytes);
+        }
+    }
+
+    /**
+     * Creates a new predictor that always returns the same prediction of
+     * the specified buffer size.
+     */
+    public NettyRecvByteBufAllocator(int bufferSize) {
+//        checkPositive(bufferSize, "bufferSize");
+        this.bufferSize = bufferSize;
+    }
+
+    @SuppressWarnings("deprecation")
+    @Override
+    public Handle newHandle() {
+        return new NettyRecvByteBufAllocator.HandleImpl(bufferSize);
+    }
+
+    @Override
+    public NettyRecvByteBufAllocator respectMaybeMoreData(boolean respectMaybeMoreData) {
+        super.respectMaybeMoreData(respectMaybeMoreData);
+        return this;
+    }
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/decoder/TransferDecoder.java

@@ -0,0 +1,79 @@
+package org.framework.lazy.cloud.network.heartbeat.common.decoder;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.wu.framework.core.exception.RuntimeExceptionFactory;
+
+import java.util.List;
+
+/**
+ * ByteBuf 解码为 byte[]
+ */
+@Slf4j
+public class TransferDecoder extends
+        ByteToMessageDecoder {
+
+    // 上一次可以读取到到数据长度
+    private long latestReadableBytes = -1;
+
+    private final int maxFrameLength; // 读取最大长度超出会异常
+    private final int perPackageLimitLength;// 粘包时合并后到包最大大小
+
+    public TransferDecoder(int maxFrameLength, int perPackageLimitLength) {
+        this.maxFrameLength = maxFrameLength;
+        this.perPackageLimitLength = perPackageLimitLength;
+    }
+
+    protected NettyByteBuf decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
+        if (in == null) {
+            log.info("this byteBuf is null");
+            return null;
+        }
+//        if (latestReadableBytes == -1) {
+//
+//            latestReadableBytes = in.readableBytes();
+//            // 第一次什么也不做
+//            return null;
+//        }
+//        // 第二次了哦
+//        if (in.readableBytes() > latestReadableBytes && in.readableBytes() < perPackageLimitLength) {
+//            // 继续粘包
+//            latestReadableBytes = in.readableBytes();
+//            log.info("粘包ing:{}",latestReadableBytes);
+//            return null;
+//        }
+//        if (in.readableBytes() > maxFrameLength) {
+//            RuntimeExceptionFactory.of("readableBytes is max then maxFrameLength:" + maxFrameLength);
+//        }
+//        log.info("粘包结束开始处理数据:{}",latestReadableBytes);
+
+        byte[] bytes = new byte[in.readableBytes()];
+        in.readBytes(bytes);
+        NettyByteBuf nettyByteBuf = new NettyByteBuf();
+        nettyByteBuf.setData(bytes);
+        latestReadableBytes = -1;
+
+        return nettyByteBuf;
+    }
+
+    /**
+     * Decode the from one {@link ByteBuf} to an other. This method will be called till either the input
+     * {@link ByteBuf} has nothing to read when return from this method or till nothing was read from the input
+     * {@link ByteBuf}.
+     *
+     * @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
+     * @param in  the {@link ByteBuf} from which to read data
+     * @param out the {@link List} to which decoded messages should be added
+     * @throws Exception is thrown if an error occurs
+     */
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        Object decoded = decode(ctx, in);
+        if (decoded != null) {
+            out.add(decoded);
+        }
+    }
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/encoder/TransferEncoder.java

@@ -0,0 +1,19 @@
+package org.framework.lazy.cloud.network.heartbeat.common.encoder;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+
+/**
+ * 编码
+ */
+public class TransferEncoder extends MessageToByteEncoder<NettyByteBuf> {
+
+
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext,NettyByteBuf nettyByteBuf, ByteBuf byteBuf) throws Exception {
+        byte[] bytes = nettyByteBuf.getData();
+        byteBuf.writeBytes(bytes);
+    }
+}