使用Netty对POJO对象进行序列化

傷城~ 2022-05-13 10:04 261阅读 0赞

使用JDK的ObjectInputStream和ObjectOutputStream可以实现java对象的序列化和反序列化（只要被序列化的POJO对象实现Serializable接口）。

在不需要考虑跨语言，并且对序列化的性能要去不苛刻时，JDK默认的序列化机制是最明智的选择之一。

下面的例子中，我们使用Netty的ObjectDecoder和ObjectEncoder对请求和应答对象进行序列化。  
  
请求对象：

/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
    public class SubscribeReq implements Serializable { 
        private int subReqId;
        private String userName;
        private String productName;
        public SubscribeReq() {
        }
        public SubscribeReq(int subReqId, String userName, String productName) {
            this.subReqId = subReqId;
            this.userName = userName;
            this.productName = productName;
        }
        public int getSubReqId() {
            return subReqId;
        }
        public void setSubReqId(int subReqId) {
            this.subReqId = subReqId;
        }
        public String getUserName() {
            return userName;
        }
        public void setUserName(String userName) {
            this.userName = userName;
        }
        public String getProductName() {
            return productName;
        }
        public void setProductName(String productName) {
            this.productName = productName;
        }
        @Override
        public String toString() {
            return "SubscribeReq{" +
                    "subReqId=" + subReqId +
                    ", userName='" + userName + '\'' +
                    ", productName='" + productName + '\'' +
                    '}';
        }
    }

应答对象：

/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
    public class SubscribeResp implements Serializable { 
        private int subReqId;
        private int respCode;
        private String desc;
        public SubscribeResp() {
        }
        public SubscribeResp(int subReqId, int respCode) {
            this.subReqId = subReqId;
            this.respCode = respCode;
        }
        public SubscribeResp(int subReqId, int respCode, String desc) {
            this.subReqId = subReqId;
            this.respCode = respCode;
            this.desc = desc;
        }
        @Override
        public String toString() {
            return "SubscribeResp{" +
                    "subReqId=" + subReqId +
                    ", respCode=" + respCode +
                    ", desc='" + desc + '\'' +
                    '}';
        }
    }

服务端  
在ChannelPipeline中增加ObjectDecoder解码器和ObjectEncoder编码器，并对要进行java序列化的对象实现Serialzable接口。

/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
    public class SubReqServer { 
        public static void main(String[] args) {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,1024).childHandler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                    socketChannel.pipeline().addLast(new ObjectEncoder());
                    socketChannel.pipeline().addLast(new SubReqServerHandler());
                }
            });
            try {
                ChannelFuture f = b.bind(9988).sync();
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }
    
    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 {
            SubscribeReq subscribeReq = (SubscribeReq) msg;
            System.out.println("接收到客户端请求：" + subscribeReq);
            ctx.writeAndFlush(resp(subscribeReq.getSubReqId()));
        }
        private SubscribeResp resp(int subReqId) {
            SubscribeResp subscribeResp = new SubscribeResp(subReqId,0);
            return subscribeResp;
        }
    }

客户端

/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
    public class SubReqClient { 
        public static void main(String[] args) {
            EventLoopGroup group = new NioEventLoopGroup();
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                    socketChannel.pipeline().addLast(new ObjectEncoder());
                    socketChannel.pipeline().addLast(new SubReqClientHandler());
                }
            });
            try {
                ChannelFuture f = b.connect("127.0.0.1",9988).sync();
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                group.shutdownGracefully();
            }
        }
    }
    class SubReqClientHandler extends ChannelHandlerAdapter {
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // 链路激活后，向服务端发送10条订购信息
            for (int i=0;i<10;i++) {
                SubscribeReq subscribeReq = new SubscribeReq(i,"tommy","Netty权威指南");
                ctx.write(subscribeReq);
            }
            ctx.flush();
        }
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            SubscribeResp subscribeResp = (SubscribeResp) msg;
            System.out.println("接收到服务端响应："  + subscribeResp);
        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    }

运行结果：  
服务端  
![76598766.jpg][]  
客户端  
![27651857.jpg][]

ObjectDecoder负责对实现了Serialzable接口的POJO对象进行解码，它有多个构造函数，支持不同的ClassResolver。  
服务端使用的是weakCachingConcurrentResolver，创建线程安全的WeakReferenceMap对类加载器进行缓存，它支持多线程并发访问。当虚拟机内存不足时，会释放缓存中的内存。  
客户端使用的是cacheDisabled，禁止对类加载器进行缓存，它在基于OSGi的动态模块化编程中经常使用。由于OSGi的bundle可以进行热部署和热升级，当某个bundle升级后，它对应的类加载器也将一起升级，因此在动态模块化编程中，很少对类加载器进行缓存，因为它随时可能发生变化。  
为了防止异常码流和解码错位导致的内存溢出，这里将单个对象的最大序列化后的字节数设置为1M。

ObjectEncoder负责将实现了Serialzable接口的POJO对象进行编码，用户对对象手动进行序列化，关注于自己的业务即可。  
对象的序列化和反序列化都由Netty的ObjectDecoder和ObjectEncoder来搞定。

参考《Netty权威指南》

[76598766.jpg]: http://omh46px9n.bkt.clouddn.com/17-11-18/76598766.jpg
[27651857.jpg]: http://omh46px9n.bkt.clouddn.com/17-11-18/27651857.jpg