netty编解码之java原生序列化

Posted 2022-08-22 叶长风

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了netty编解码之java原生序列化相关的知识，希望对你有一定的参考价值。

netty序列化之java原生序列化

前几天有空看了下netty使用java原生序列化、以及使用protobuf、jboss marshalling进行编解码，各种技术之间差异挺大，使用的方法各自不同，性能上原生的性能优势确实不大，但是另外两种确实很有优势，觉得有点意思，于是准备写一篇文章记载一下，但是一篇文章篇幅太长，因此拆成了三篇文章分别来讲。

java序列化代码

这个简单的程序就是server和client互相通信，因此程序就是以下几个类，比较简单，本文采用netty4来进行编程。

model类

请求model类SubscribeReq:

package cn.com.serialize;

import java.io.Serializable;

/**
 * Created by xiaxuan on 17/11/27.
 */
public class SubscribeReq implements Serializable 

    private int subReqID;
    private String userName;
    private String productName;
    private String phoneNumber;
    private String address;

    ....

中间get、set方法省略，以下SubscribeResp也是如此。

响应model类SubscribeResp:

package cn.com.serialize;

import java.io.Serializable;

/**
 * Created by xiaxuan on 17/11/27.
 */
public class SubscribeResp implements Serializable 

    private int subReqID;
    private int respCode;
    private String desc;

    ....

server类SubReqServer和client类SubReqClient

server类SubReqServer

package cn.com.serialize;

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;

/**
 * Created by xiaxuan on 17/11/27.
 */
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 ch) throws Exception 
                            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();
         catch (InterruptedException e) 
            e.printStackTrace();
         finally 
            //优雅退出，释放线程组资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        
    

    public static void main(String[] args) 
        int port = 8080;
        new SubReqServer().bind(port);

client类SubReqClient

package cn.com.serialize;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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;

/**
 * Created by xiaxuan on 17/11/27.
 */
public class SubReqClient 

    public void connect(int port, String host) 
        //配置客户端NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try 
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected 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();
         catch (InterruptedException e) 
            e.printStackTrace();
         finally 
            //优雅退出，释放NIO线程组
            group.shutdownGracefully();
        
    

    public static void main(String[] args) 
        int port = 8080;
        new SubReqClient().connect(port, "127.0.0.1");

自己编写java的序列化太过麻烦，但是netty中提供两个类ObjectDecoder和ObjectEncoder来进行进行序列化。

ObjectDecoder有多种构造函数，支持不同的ClassResolver，在此使用了weakCachingConcurrentResolver创建线程安全的WeakReferenceMap对类加载器进行缓存，并且当虚拟机内存不足的时候，会释放缓存中的内存，防止内存泄漏，为了防止异常码流和解码错位导致的内存溢出，这里将单个对象最大序列化后的字节数组长度设置为1M，但是作为本例其实已经足够了。

在server类和client类在启动代码编写完了后，都会有finally块中将两个EventLoopGroup优雅关闭的代码，这里实际上是进行了hock的注册，在程序直接关闭的时候并不是直接杀死程序，而是接收到kill事件之后一步步关闭程序，这个原理以后有空再单独写一篇文章讲一下。

客户端业务逻辑处理类SubReqClientHandler:

package cn.com.serialize;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * Created by xiaxuan on 17/11/27.
 */
public class SubReqClientHandler extends ChannelInboundHandlerAdapter 

    public SubReqClientHandler() 

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception 
        for (int i = 0; i < 10; i++) 
            ctx.write(subReq(i));
        
        ctx.flush();
    

    private SubscribeReq subReq(int i) 
        SubscribeReq req = new SubscribeReq();
        req.setAddress("北京市");
        req.setPhoneNumber("135xxxxxxxx");
        req.setProductName("netty书籍");
        req.setUserName("xiaxuan");
        req.setSubReqID(i);
        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) throws Exception 
        cause.printStackTrace();
        ctx.close();

客户端链路被激活时构造订购请求消息发送，客户端一次构造十条订购请求，最后一次性发送给服务端，然后打出服务端的响应。

server业务逻辑处理类SubReqServerHandler:

package cn.com.serialize;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * Created by xiaxuan on 17/11/27.
 */
@ChannelHandler.Sharable
public class SubReqServerHandler extends ChannelInboundHandlerAdapter 

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        SubscribeReq req = (SubscribeReq) msg;
        if ("xiaxuan".equalsIgnoreCase(req.getUserName())) 
            System.out.println("Service accept client subscribe 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) throws Exception 
        cause.printStackTrace();
        ctx.close();