Netty序列化深入理解与使用

简介

Netty是基于NIO实现的Java网络编程框架，它也是目前使用最为广泛的Java网络编程框架之一。在进行远程通信时，往往需要将对象进行序列化、反序列化后再进行传输，Netty自带了一些序列化方式，比如JDK序列化、Protobuf等，使用Netty内置的序列化方式可以方便地实现对象的传输。本文将详细讲解Netty序列化的使用、原理以及如何自定义序列化方式。

Netty内置的序列化方式

JDK序列化

JDK序列化是最简单、最常用的序列化方式之一。在序列化时，Java对象会被转换为二进制流，可以在网络中进行传输，收到数据后再将二进制流转换为Java对象。使用JDK序列化时，需要将对象类实现Serializable接口。

public class User implements Serializable {

    private String name;
    private int age;
    private List<String> hobbies;

    // Getter 和 Setter 省略
}

在Netty中使用JDK序列化，需要在ChannelPipeline中加入后续的处理器(如编码器和解码器)。

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new ObjectEncoder());
        pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
        pipeline.addLast(new MyServerHandler());
    }
}

Protobuf序列化

Protobuf是谷歌公司开源的一种高效的序列化协议。相比于JDK序列化，Protobuf序列化更加高效，可读性更强，编写方式也更加友好。使用Protobuf序列化，需要定义.proto文件并使用插件生成对应的Java文件。

proto文件示例：

syntax = "proto3";

message User {
    string name = 1;
    int32 age = 2;
    repeated string hobbies = 3;
}

生成的Java文件中包含了定义的message类。

在Netty中使用Protobuf序列化需要进行两个步骤：

1. 定义将要序列化的类以及对应的.proto文件
2. 在ChannelPipeline中加入后续的处理器(如编码器和解码器)

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        // 定义将要序列化的类以及对应的proto文件路径
        pipeline.addLast(new ProtobufEncoder());
        pipeline.addLast(new ProtobufDecoder(UserProto.User.getDefaultInstance()));

        pipeline.addLast(new MyServerHandler());
    }
}

自定义序列化方式

Netty提供了自定义序列化方式的接口，用户可以根据自己的需求编写自己的序列化、反序列化的逻辑。

编写Encoder

我们需要编写一个实现了MessageToByteEncoder接口的编码器，将Java对象转换为二进制数据。

public class CustomEncoder extends MessageToByteEncoder<User> {

    @Override
    protected void encode(ChannelHandlerContext ctx, User msg, ByteBuf out) throws Exception {
        byte[] nameBytes = msg.getName().getBytes(Charset.forName("utf-8"));
        out.writeInt(nameBytes.length);
        out.writeBytes(nameBytes);
        out.writeInt(msg.getAge());
        out.writeInt(msg.getHobbies().size());
        for (String hobby : msg.getHobbies()) {
            byte[] hobbyBytes = hobby.getBytes(Charset.forName("utf-8"));
            out.writeInt(hobbyBytes.length);
            out.writeBytes(hobbyBytes);
        }
    }
}

编写Decoder

我们需要编写一个实现了ByteToMessageDecoder接口的解码器，将二进制数据还原为Java对象。

public class CustomDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int nameLength = in.readInt();
        byte[] nameBytes = new byte[nameLength];
        in.readBytes(nameBytes);
        String name = new String(nameBytes, Charset.forName("utf-8"));

        int age = in.readInt();

        int hobbyCount = in.readInt();
        List<String> hobbies = new ArrayList<>(hobbyCount);
        for (int i = 0; i < hobbyCount; i++) {
            int hobbyLength = in.readInt();
            byte[] hobbyBytes = new byte[hobbyLength];
            in.readBytes(hobbyBytes);
            hobbies.add(new String(hobbyBytes, Charset.forName("utf-8")));
        }

        out.add(new User(name, age, hobbies));
    }
}

使用自定义的序列化方式

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        // 使用我们上面编写的自定义Encoder和Decoder
        pipeline.addLast(new CustomEncoder());
        pipeline.addLast(new CustomDecoder());

        pipeline.addLast(new MyServerHandler());
    }
}

示例

以下是一个简单的使用自定义序列化方式的示例。

1. 客户端发送User对象

public class MyClient {

    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new MyClientInitializer());

            ChannelFuture f = b.connect("localhost", 8888).sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

public class MyClientHandler extends SimpleChannelInboundHandler<User> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        List<String> hobbies = new ArrayList<>();
        hobbies.add("reading");
        hobbies.add("coding");
        ctx.writeAndFlush(new User("Alice", 18, hobbies));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, User msg) throws Exception {
        // do nothing
    }
}

public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        // 使用自定义编码器和解码器
        pipeline.addLast(new CustomEncoder());
        pipeline.addLast(new CustomDecoder());

        pipeline.addLast(new MyClientHandler());
    }
}

2. 服务端接收User对象

public class MyServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new MyServerInitializer())
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(8888).sync();
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

public class MyServerHandler extends SimpleChannelInboundHandler<User> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, User msg) throws Exception {
        System.out.println("Received user: " + msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

总结

以上是Netty序列化深入理解与使用的完整攻略。本文讲解了Netty内置的序列化方式、自定义序列化方式的实现以及两个示例。希望能够对读者对Netty序列化有更深的理解和掌握。