netty 4.1.45自定义编解码器

逃离我推掉我的手 2023-07-04 06:17 39阅读 0赞

### 文章目录 ###

*  概述
 *  自定义编解码器实现机制
 *  编码器
 *  解码器
 *  服务端代码
 *  客户端代码
 *  运行结果

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

# 概述 #

前面 [netty 4.1.45 TCP拆包粘包原因及解决办法][netty 4.1.45 TCP] 介绍了使用netty本身提供的LineBasedFrameDecoder、DelimiterBasedFrameDecoder、FixedLengthFrameDecoder三种编解码方案。不过实际工作中直接使用这几种编解码的还是比较少的。比如处于安全、性能、私有网络等。

这里介绍简单的自定义编解码器的实现，后续再介绍如何与json，PB协议等进行集成实现复杂的netty协议。

# 自定义编解码器实现机制 #

将消息分为`消息头和消息体`，消息头用int类型表示，当然你们也可以用其他类型标识。消息头的值是消息体的长度。  
编码时，首先读取消息的长度，写入发送缓冲区，再将消息体写入；  
解码时，要先读取消息的长度，然后读取该长度的消息体，当拆包或者粘包时要将数据报进行拆分或者合并。

我们发送的每一条数据的格式：  
4字节的长度+实际数据内容  
如：10helloworld，表示长度是10，从后读取10个字节ji是一条完整的消息。

# 编码器 #

编码器需要实现MessageToByteEncoder类，该类用途就是读取数据长度，分两次写入缓冲区，一次数据长度，一次消息体。

public class MyCustomMessageEncoder extends MessageToByteEncoder<Object> { 
        @Override
        protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) { 
            // 要发送的数据
            // 这里如果是自定义的类型，msg即为自定义的类型，需要转为byte[]
            byte[] body = ((ByteBuf) msg).array();
    
            // 数据长度
            int dataLength = body.length;
            // 缓冲区先写入数据长度
            out.writeInt(dataLength);
            // 再写入数据
            out.writeBytes(body);
        }
    }

# 解码器 #

public class MyCustomMessageDecoder extends ByteToMessageDecoder { 
        // 消息头：发送端写的是一个int，占用4字节。
        private final static int HEAD_LENGTH = 4;
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
            //
            if (in.readableBytes() < HEAD_LENGTH) { 
                return;
            }
    
            // 标记一下当前的readIndex的位置
            in.markReaderIndex();
    
            // 读取数据长度
            int dataLength = in.readInt();
            // 我们读到的消息体长度为0，这是不应该出现的情况，这里出现这情况，关闭连接。
            if (dataLength < 0) { 
                System.out.println("ERROR!");
                ctx.close();
            }
    
            //读到的消息体长度如果小于我们传送过来的消息长度，则resetReaderIndex.
            // 这个配合markReaderIndex使用的。
            // 把readIndex重置到mark的地方
            if (in.readableBytes() < dataLength) { 
                in.resetReaderIndex();
                return;
            }
    
            // 将缓冲区的数据读到字节数组
            byte[] body = new byte[dataLength];
            in.readBytes(body);
            //将byte数据转化为我们需要的对象。
            Object msg = convertToObj(body);
            out.add(msg);
        }
    
        private Object convertToObj(byte[] body) { 
            return new String(body, 0, body.length);
        }
    }

这里需要着重注意点是ByteBuf的`markReaderIndex`和`resetReaderIndex`的配合使用。

*  `markReaderIndex`：标记一下当前读的位置
 *  `resetReaderIndex`：重置读取位置到标记的位置  
    从代码中也可以看到，如果可读的字节数小于数据长度dataLength，需要等待下一个包到来继续读取。

# 服务端代码 #

服务端最重要的是配置channelpipeline。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]  
完整代码如下：

package com.example;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.ByteToMessageDecoder;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.MessageToByteEncoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.List;
    
    /** * 自定义消息解码器 */
    public class MyNettyServer4 { 
        public static void main(String[] args) { 
            //配置服务端NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
    
            try { 
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup)//配置主从线程组
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)//配置一些TCP的参数
                        .childHandler(new MyChildHandler4());//添加自定义的channel
                //绑定8080端口
                ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
                //服务端监听端口关闭
                ChannelFuture future = channelFuture.channel().closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                //netty优雅停机
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
    }
    
    class MyChildHandler4 extends ChannelInitializer<SocketChannel> { 
        @Override
        protected void initChannel(SocketChannel socketChannel) { 
            socketChannel.pipeline()
                    .addLast(new MyCustomMessageEncoder())
                    .addLast(new MyCustomMessageDecoder())
                    .addLast(new TimeServerHandler4());
        }
    }
    
    /** * TimeServerHandler这个才是服务端真正处理请求的服务方法 */
    class TimeServerHandler4 extends ChannelInboundHandlerAdapter { 
    
        private int count = 0;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            String body = (String) msg;
    
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String currentTimeStr = "QUERY TIME ORDER".equalsIgnoreCase(body) ?
                    format.format(new Date()) + "" : "BAD ORDER";
    
            //受到一次请求，count加1
            System.out.println("第 " + count++ + " 次收到客户端请求：" + body + " 返回响应：" + currentTimeStr);
    
            ByteBuf resp = Unpooled.copiedBuffer(currentTimeStr.getBytes());
    
            ctx.write(resp);//将消息发送到发送缓冲区
    // ctx.writeAndFlush(resp);//如果这里使用writeAndFlush，则下面channelReadComplete中就不需要flush了
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) { 
            ctx.flush();//将消息从发送缓冲区中写入socketchannel中
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            ctx.close();
        }
    }

# 客户端代码 #

客户端代码最重要也是配置channelpipeline。

完整代码：

package com.example;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.*;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    /** * 自定义消息解码器 */
    public class MyNettyClient4 { 
        public static void main(String[] args) { 
            //客户端NIO线程组
            EventLoopGroup group = new NioEventLoopGroup();
    
            try { 
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel ch) { 
                                ch.pipeline()
                                        .addLast(new MyCustomMessageEncoder())
                                        .addLast(new MyCustomMessageDecoder())
                                        .addLast(new TimeClientHandler4());
    
                            }
                        });
    
                //异步链接服务器
                ChannelFuture future = bootstrap.connect("127.0.0.1", 8080).sync();
                //等等客户端链接关闭
                future.channel().closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                //优雅停机
                group.shutdownGracefully();
            }
        }
    }
    
    //客户端业务逻辑处理类
    class TimeClientHandler4 extends ChannelInboundHandlerAdapter { 
        private int count = 0;
    
        /** * 客户端与服务器TCP链路链接成功后调用该方法 * * @param ctx */
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            byte[] req = ("QUERY TIME ORDER").getBytes();
            for (int i = 0; i < 100; i++) { 
                ByteBuf firstMsg = Unpooled.buffer(req.length);
                firstMsg.writeBytes(req);
                ctx.writeAndFlush(firstMsg);//写入缓冲并且发送到socketchannel
            }
        }
    
        /** * 读取到服务端相应后执行该方法 * * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) { 
            String body = (String) msg;
            System.out.println("第 " + count++ + " 次受到服务端返回：" + body);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            System.out.println("Unexpected exception from downstream : " + cause.getMessage());
            ctx.close();
        }
    }

# 运行结果 #

server端结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]

客户端结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]  
结果正常。

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

[netty]: https://blog.csdn.net/fgyibupi/category_9705823.html
[netty 4.1.45 TCP]: https://it007.blog.csdn.net/article/details/104224209
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200208214427685.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200208214118908.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200208214049590.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70