Netty(三)——编解码之路

冷不防 2022-05-03 04:26 325阅读 0赞

我们在开发中经常会把Java类进行implements Serializable用来网络传输的序列化和反序列化，过程其实就是将Java对象转编码为字节数组或者ByteBuffer对象进行传输，当远程服务读取到ByteBuffer对象或者字节数组时，需要将其解码为Java对象。这也就是编解码技术。

Java的序列化只是编解码技术中的一种，但是由于其1，不支持跨语言；2，序列化后的码流相对较大（Java.io.Serializable序列化相对于基于BuyeBuffer的二进制编码的码流大小大概为5：1）；3，性能相对来说较差（Java序列化只有二进制编码的大概6%）。业界出现了很多编解码的框架，我们来看看我们选择一款编解码框架时需要考虑的东西：

1，是否支持跨语言，支持的语言种类是否丰富（现在跨系统、跨语言的交互太多了）；

2，编码后的码流大小（码流越大，存储占空间、网络传输更占带宽、吞吐量越低）；

3，编解码的性能；

4，类库是否小巧，API使用是否方便；

5，使用者需要手工开发的工作量和难度。

好，下边讲述三种业界比较流行，而且Netty支持，通过Netty编程来实现的编解码框架；

一，MessagePack编解码：MessagePack is an efficient binary serialization format. It lets you exchange data among multiple languages like JSON. But it's faster and smaller. Small integers are encoded into a single byte, and typical short strings require only one extra byte in addition to the strings themselves.（官方解释）。它具有a，编解码高效、性能高；b，序列化之后码流小；c，支持跨语言（Java、Python、Ruby、C\#、Lua、Go、C、C++……）

好，看下MessagePack编解码的几个重要点：1，利用MessagePack实现编码和解码的ChannelHandlerAdapter；2，server和client端，initChannel时将编解码ChannelHandler添加到pipeline；3，支持粘包半包操作：LengthFieldBasedFrameDecoder、LengthFieldPrepender的加入。重点代码如下：

/**
     * @author liujiahan
     * @Title: MsgpackDecoder
     * @Copyright: Copyright (c) 2018
     * @Description: 利用MessagePack解码 MessageToMessageDecoder<I> extends ChannelHandlerAdapter
     * @Created on 2018/10/24
     * @ModifiedBy:
     */
    public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf> {
    
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
            final byte[] array;
            final int length = byteBuf.readableBytes();
            array = new byte[length];
            byteBuf.getBytes(byteBuf.readerIndex(),array,0,length);
            MessagePack messagePack = new MessagePack();
            list.add(messagePack.read(array));
        }
    }
    
    
    
    /**
     * @author liujiahan
     * @Title: MsgpackEncoder
     * @Copyright: Copyright (c) 2018
     * @Description: 利用MessagePack编码 MessageToByteEncoder<I> extends ChannelHandlerAdapter
     * @Created on 2018/10/24
     * @ModifiedBy:
     */
    public class MsgpackEncoder extends MessageToByteEncoder<Object> {
    
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
            MessagePack messagePack =new MessagePack();
            //Serialize
            byte[] raw = messagePack.write(o);
            byteBuf.writeBytes(raw);
    
        }
    }
    
    
    /**
    *  server和client端channel handler
    */
    Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //处理解决粘包半包
                        socketChannel.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
                        //解码
                        socketChannel.pipeline().addLast("msgpack decoder",new MsgpackDecoder());
                        //处理解决粘包半包
                        socketChannel.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                        //编码
                        socketChannel.pipeline().addLast("msgpack encoder",new MsgpackEncoder());
                        socketChannel.pipeline().addLast(new EchoClientHandler());
                    }
                });
                ChannelFuture f = b.connect(host, port).sync();
                f.channel().closeFuture().sync();

二，Google Protobuf编解码（https://developers.google.com/protocol-buffers/docs/javatutorial）：它的优点a，在谷歌长期使用，产品成熟度高；b，跨语言、支持多种语言，包括C++、Java和Python等；c，编码后的消息更小，更加有利于存储和传输；d，编解码的性能非常高；e，支持不同协议版本的前向兼容；f，支持定义可选和必选字段。

首先，1，需要安装protobuf；2，编写proto数据文件；3，通过protoc的命令（通过protoc -help看命令，这里看下生成Java的：  --java\_out=OUT\_DIR          Generate Java source file.）进行生成对应语言的类 ；4，编写代码。

这个例子看下全的代码：

1，首先是两个request和response的proto数据文件的定义，然后通过protoc生成对应的java实体，代码多不展示实体代码。

package netty;
    option java_package = "com.ljh.netty.protobuf";
    option java_outer_classname = "SubscribeReqProto";
    
    message SubscribeReq{
        required int32 subReqId = 1;
        required string userName = 2;
        required string productName = 3;
        required string address = 4;
    
    }
    
    
    
    package netty;
    option java_package = "com.ljh.netty.protobuf";
    option java_outer_classname = "SubscribeRespProto";
    
    message SubscribeResp{
        required int32 subReqId = 1;
        required int32 subRespCode = 2;
        required string desc = 3;
    }

2，server和client端代码：

public class SubReqServer {
    
        public void bind(int port){
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap b =new ServerBootstrap();
                b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,100).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
    
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                        socketChannel.pipeline().addLast(new ProtobufDecoder(SubscribeReqProto.SubscribeReq.getDefaultInstance()));
                        socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                        socketChannel.pipeline().addLast(new ProtobufEncoder());
                        socketChannel.pipeline().addLast(new SubReqServerHandler());
                    }
                });
                ChannelFuture f =b.bind(port).sync();
                f.channel().closeFuture().sync();
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) {
            int port =8091;
            if(args !=null && args.length >0){
                try {
                    port = Integer.valueOf(args[0]);
                }catch (NumberFormatException e){
    
                }
            }
    
            new SubReqServer().bind(port);
        }
    }
    
    
    
    
    public class SubReqServerHandler extends ChannelHandlerAdapter {
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            SubscribeReqProto.SubscribeReq req = (SubscribeReqProto.SubscribeReq) msg;
            if ("ljh".equalsIgnoreCase(req.getUserName())) {
                System.out.println("server accept client subscribe req : [" + req.toString() + "]");
                ctx.writeAndFlush(resp(req.getSubReqId()));
            }
        }
    
        private SubscribeRespProto.SubscribeResp resp(int subReqId) {
            SubscribeRespProto.SubscribeResp.Builder builder = SubscribeRespProto.SubscribeResp.newBuilder();
            builder.setSubReqId(subReqId);
            builder.setSubRespCode(0);
            builder.setDesc("netty book is very good");
            return builder.build();
    
        }
    }
    
    
    
    public class SubReqClient {
        public void connect(int port,String host){
            EventLoopGroup group = new NioEventLoopGroup();
            try{
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //操，搞了我一个小时
                        socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                        socketChannel.pipeline().addLast(new ProtobufDecoder(SubscribeRespProto.SubscribeResp.getDefaultInstance()));
                        socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                        socketChannel.pipeline().addLast(new ProtobufEncoder());
                        socketChannel.pipeline().addLast(new SubReqClientHandle());
                    }
                });
                ChannelFuture f = b.connect(host,port).sync();
                f.channel().closeFuture().sync();
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) {
            int port =8091;
            if(args !=null && args.length >0){
                try {
                    port = Integer.valueOf(args[0]);
                }catch (NumberFormatException e){
    
                }
            }
    
            new SubReqClient().connect(port,"127.0.0.1");
        }
    }
    
    
    
    public class SubReqClientHandle extends ChannelHandlerAdapter {
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            for(int i = 0 ;i<10 ;i++){
                ctx.write(subReq(1));
            }
            ctx.flush();
        }
    
        private SubscribeReqProto.SubscribeReq subReq(int i){
            SubscribeReqProto.SubscribeReq.Builder builder = SubscribeReqProto.SubscribeReq.newBuilder();
            builder.setSubReqId(i);
            builder.setUserName("ljh");
            builder.setProductName("netty-book");
            builder.setAddress("beijing");
            return builder.build();
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("receive server response ：【" + msg + "】");
        }
    }

三，Jboss Marshalling编解码(http://jbossmarshalling.jboss.org/)：也是一个Java对象序列包，对JDK默认的序列化框架进行了优化，又保持了跟Java.io.Serializable接口的兼容，同时增加了一些可调的参数和附加的特性，这些参数和特性可通过工厂类进行配置。

编码的重点看下吧，1，通过MarshallerFactory、MarshallingConfiguration创建UnmarshallerProvider实例，并创建MarshallingDecoder的channelHandler和MarshallingEncoder的channelHandler。2，server和client的initChannel时将编解码ChannelHandler添加到pipeline。

好，下边看下重点代码：

/**
     * @author liujiahan
     * @Title: MarshallingCodeFactory
     * @Copyright: Copyright (c) 2018
     * @Description:  MarshallingCodeFactory 通过marshallerFactory 创建MarshallingDecoder decoder,MarshallingEncoder encoder
     * @Created on 2018/10/26
     * @ModifiedBy:
     */
    public final class MarshallingCodeFactory {
    
        /**
         * 创建JBoss Marshalling解码器MarshallingDecoder
         *
         * @return
         */
        public static MarshallingDecoder buildMarshallingDecoder() {
            final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration = new MarshallingConfiguration();
            configuration.setVersion(5);
            UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
            MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024);
            return decoder;
        }
    
        /**
         * 创建 JBoss Marshalling编码器MarshallingEncoder
         *
         * @return
         */
        public static MarshallingEncoder buildMarshallingEncoder() {
            final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration = new MarshallingConfiguration();
            configuration.setVersion(5);
            MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
            MarshallingEncoder encoder = new MarshallingEncoder(provider);
            return encoder;
        }
    }
    
    
    
    //片段
    ServerBootstrap b =new ServerBootstrap();
                b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,100).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
    
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //解码
                        socketChannel.pipeline().addLast(MarshallingCodeFactory.buildMarshallingDecoder());
                        //编码
                        socketChannel.pipeline().addLast(MarshallingCodeFactory.buildMarshallingEncoder());
                        socketChannel.pipeline().addLast(new SubReqServerHandler());
                    }
                });

好，Netty支持很多成熟的高效的编解码框架，用来解决网络编程的编解码问题非常方便。

其实经过几个例子后，就会发现、总结出Netty编程的套路，慢慢就能找到它的门道，接下来会继续学习，相信也会理解更加深刻。