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

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

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

请求对象:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
/**
 * @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 + '\'' +
                '}';
    }
}

应答对象:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
 * @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接口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
/**
 * @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;
    }
}

客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
/**
 * @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();
    }
}

运行结果:
服务端

客户端

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

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

参考《Netty权威指南》