Netty之编解码开发（五）

深藏阁楼爱情的钟 2023-03-04 13:26 132阅读 0赞

### 文章目录 ###

*   *  分析
     *  分析：
     *  业界主流的编解码框架
     *  Netty中编解码
     *  结果
     *  其他

我们选择一个编解码框架的时，往往会考虑以下几个因素：

*  是否支持跨语言，支持的语言种类是否丰富；
 *  编码后的码流大小
 *  编解码的性能；
 *  类库是否小巧，API使用是否方便
 *  使用者需要手工开发的工作量和难度。

package main.com.phei.netty.codec.serializable;
    
    import java.io.Serializable;
    import java.nio.ByteBuffer;
    
    /** * @author Administrator * @version 1.0 * @date 2014年2月23日 */
    public class UserInfo implements Serializable { 
    
        /** * 默认的序列号 */
        private static final long serialVersionUID = 1L;
    
        private String userName;
    
        private int userID;
    
        public UserInfo buildUserName(String userName) { 
            this.userName = userName;
            return this;
        }
    
        public UserInfo buildUserID(int userID) { 
            this.userID = userID;
            return this;
        }
    
        /** * @return the userName */
        public final String getUserName() { 
            return userName;
        }
    
        /** * @param userName the userName to set */
        public final void setUserName(String userName) { 
            this.userName = userName;
        }
    
        /** * @return the userID */
        public final int getUserID() { 
            return userID;
        }
    
        /** * @param userID the userID to set */
        public final void setUserID(int userID) { 
            this.userID = userID;
        }
    
        public byte[] codeC() { 
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            byte[] value = this.userName.getBytes();
            buffer.putInt(value.length);
            buffer.put(value);
            buffer.putInt(this.userID);
            buffer.flip();
            value = null;
            byte[] result = new byte[buffer.remaining()];
            buffer.get(result);
            return result;
        }
    
        public byte[] codeC(ByteBuffer buffer) { 
            buffer.clear();
            byte[] value = this.userName.getBytes();
            buffer.putInt(value.length);
            buffer.put(value);
            buffer.putInt(this.userID);
            buffer.flip();
            value = null;
            byte[] result = new byte[buffer.remaining()];
            buffer.get(result);
            return result;
        }
    }

package main.com.phei.netty.codec.serializable;
    
    import java.io.ByteArrayOutputStream;
    import java.io.IOException;
    import java.io.ObjectOutputStream;
    
    /** * @author Administrator * @version 1.0 * @date 2014年2月23日 */
    public class TestUserInfo { 
    
        /** * @param args * @throws IOException */
        public static void main(String[] args) throws IOException { 
            UserInfo info = new UserInfo();
            info.buildUserID(100).buildUserName("Welcome to Netty");
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream os = new ObjectOutputStream(bos);
            os.writeObject(info);
            os.flush();
            os.close();
            byte[] b = bos.toByteArray();
            System.out.println("The jdk serializable length is : " + b.length);
            bos.close();
            System.out.println("-------------------------------------");
            System.out.println("The byte array serializable length is : "
                    + info.codeC().length);
    
        }
    
    }

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]

## 分析 ##

上面是Java序列化的字节码的长度，下面是二进制编码的长度，可见Java原生的序列化的`性能很差`。

package main.com.phei.netty.codec.serializable;
    
    import java.io.ByteArrayOutputStream;
    import java.io.IOException;
    import java.io.ObjectOutputStream;
    import java.nio.ByteBuffer;
    
    /** * @author Administrator * @version 1.0 * @date 2014年2月23日 */
    public class PerformTestUserInfo { 
    
        /** * @param args * @throws IOException */
        public static void main(String[] args) throws IOException { 
            UserInfo info = new UserInfo();
            info.buildUserID(100).buildUserName("Welcome to Netty");
            int loop = 1000000;
            ByteArrayOutputStream bos = null;
            ObjectOutputStream os = null;
            long startTime = System.currentTimeMillis();
            for (int i = 0; i < loop; i++) { 
                bos = new ByteArrayOutputStream();
                os = new ObjectOutputStream(bos);
                os.writeObject(info);
                os.flush();
                os.close();
                byte[] b = bos.toByteArray();
                bos.close();
            }
            long endTime = System.currentTimeMillis();
            System.out.println("The jdk serializable cost time is : "
                    + (endTime - startTime) + " ms");
    
            System.out.println("-------------------------------------");
    
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            startTime = System.currentTimeMillis();
            for (int i = 0; i < loop; i++) { 
                byte[] b = info.codeC(buffer);
            }
            endTime = System.currentTimeMillis();
            System.out.println("The byte array serializable cost time is : "
                    + (endTime - startTime) + " ms");
    
        }
    
    }

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]

## 分析： ##

对Java序列化和二进制编码分别进行性能测试，编码100万次，然后统计耗费的总时间，可以看到序列化的`速度性能也很差`。  
`还有最重要的是Java序列化无法跨语言，也就是说，如果服务使用C++或者其他语言开发的，就无法使用Java序列化写的框架与之通信。`

## 业界主流的编解码框架 ##

Protobuf全称 Google Protocol Buffers，它由谷歌开源而来，在谷歌内部久经考验。它将数据结构以。 proto文件进行描述，通过代码生成工具可以生成对应数据结构的POJO对象和 Protobuf相关的方法和属性

时间：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]  
空间：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]

## Netty中编解码 ##

package main.com.phei.netty.codec.serializable.netty;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class SubReqClient { 
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new SubReqClient().connect(port, "127.0.0.1");
        }
    
        public void connect(int port, String host) throws Exception { 
            // 配置客户端NIO线程组
            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
                            public void initChannel(SocketChannel ch)
                                    throws Exception { 
                                ch.pipeline().addLast(
                                        new ObjectDecoder(1024, ClassResolvers
                                                .cacheDisabled(this.getClass()
                                                        .getClassLoader())));
                                ch.pipeline().addLast(new ObjectEncoder());
                                ch.pipeline().addLast(new SubReqClientHandler());
                            }
                        });
    
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
    
                // 当代客户端链路关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    }

package main.com.phei.netty.codec.serializable.netty;
    
    import main.com.phei.netty.codec.pojo.SubscribeReq;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class SubReqClientHandler extends ChannelHandlerAdapter { 
    
        /** * Creates a client-side handler. */
        public SubReqClientHandler() { 
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            for (int i = 0; i < 10; i++) { 
                ctx.write(subReq(i));
            }
            ctx.flush();
        }
    
        private SubscribeReq subReq(int i) { 
            SubscribeReq req = new SubscribeReq();
            req.setAddress("Netty实战");
            req.setPhoneNumber("138xxxxxxxxx");
            req.setProductName("Netty 最佳实践和原理分析");
            req.setSubReqID(i);
            req.setUserName("fjj");
            return req;
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            System.out.println("Receive server response : [" + msg + "]");
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            cause.printStackTrace();
            ctx.close();
        }
    }

package main.com.phei.netty.codec.serializable.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class SubReqServer { 
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new SubReqServer().bind(port);
        }
    
        public void bind(int port) throws Exception { 
            // 配置服务端的NIO线程组
            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
                            public void initChannel(SocketChannel ch) { 
                                ch.pipeline()
                                        .addLast(
                                                new ObjectDecoder(
                                                        1024 * 1024,
                                                        ClassResolvers
                                                                .weakCachingConcurrentResolver(this
                                                                        .getClass()
                                                                        .getClassLoader())));
                                ch.pipeline().addLast(new ObjectEncoder());
                                ch.pipeline().addLast(new SubReqServerHandler());
                            }
                        });
    
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package main.com.phei.netty.codec.serializable.netty;
    
    import main.com.phei.netty.codec.pojo.SubscribeReq;
    import main.com.phei.netty.codec.pojo.SubscribeResp;
    import io.netty.channel.ChannelHandler.Sharable;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    @Sharable
    public class SubReqServerHandler extends ChannelHandlerAdapter { 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            SubscribeReq req = (SubscribeReq) msg;
            if ("Lilinfeng".equalsIgnoreCase(req.getUserName())) { 
                System.out.println("Service accept client subscrib req : ["
                        + req.toString() + "]");
                ctx.writeAndFlush(resp(req.getSubReqID()));
            }
        }
    
        private SubscribeResp resp(int subReqID) { 
            SubscribeResp resp = new SubscribeResp();
            resp.setSubReqID(subReqID);
            resp.setRespCode(0);
            resp.setDesc("Netty book order succeed, 3 days later, sent to the designated address");
            return resp;
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            cause.printStackTrace();
            ctx.close();// 发生异常，关闭链路
        }
    }

服务端运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]

## 结果 ##

客户端接收到了10条订购应答消息， Netty的 Objectencoder编码器可以自动对订购应答消息进行序列化，然后发送给客户端，客户端的 Objectdecoder对码流进行反序列化，获得订购请求应答消息。  
这个在前面的章节已经演示过了（[粘包/拆包][Link 1]），

## 其他 ##

除了ObjectDecoder，ObjectEncoder，还可以使用其他序列化框架在Netty中进行开发，比如ProtobufDecoder，只要在pipeline添加ProtobufDecoder(SubscribeReqProto.SubscribeReq.getDefaultInstance())以及与之相对应的粘包拆包的ProtobufVarint32FrameDecoder;  
或者使用Marshalling进行编解码操作，只要构建出MarshallingDecoder和MarshallingEncoder对象然后添加到pipeline中即可，Marshalling编解码器是支持粘包拆包的处理的，所以只需要将Marshalling添加即可。可见Netty的开发是很方便的。

构建MarshallingDecoder代码：

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);

构建MarshallingEncoder代码：

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);

> [Netty之Java I/O演进之路（一）][Netty_Java I_O]  
> [Netty之NIO入门（二）][Netty_NIO]  
> [Netty之Netty入门应用（三）][Netty_Netty]  
> [Netty之TCP粘包/拆包问题（四）][Link 1]  
> [Netty之编解码开发（五）][Netty]  
> [Netty之ChannelPipeline(六)][Netty_ChannelPipeline]  
> [Netty之EventLoop和EventLoopGroup(七)][Netty_EventLoop_EventLoopGroup]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]: /images/20230208/31e9734325a24f38839f999d3dd16536.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]: /images/20230208/369484c7674b444a8b8fb1867a7333bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]: /images/20230208/170f4c77799844128b333830fc5550e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]: /images/20230208/0d184c2f305644bfa1aecc2fee70c3f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]: /images/20230208/9f4ae42af92a4cbea8338b93422d02c4.png
[Link 1]: https://blog.csdn.net/jianjun_fei/article/details/107563110
[Netty_Java I_O]: https://blog.csdn.net/jianjun_fei/article/details/107505729
[Netty_NIO]: https://blog.csdn.net/jianjun_fei/article/details/107513219
[Netty_Netty]: https://blog.csdn.net/jianjun_fei/article/details/107555647
[Netty]: https://blog.csdn.net/jianjun_fei/article/details/107772642
[Netty_ChannelPipeline]: https://blog.csdn.net/jianjun_fei/article/details/107787864
[Netty_EventLoop_EventLoopGroup]: https://blog.csdn.net/jianjun_fei/article/details/107956949