Netty学习二-编解码技术

1 编解码简介

编解码技术，说白了就是java序列化技术，可以持久化对象或网络传输。虽然可以使用java进行对象序列化，netty去传输，但是java序列化的硬伤太多，比如java序列化没法跨语言、序列化后码流量太大、序列化性能太低等等；主流的编解码框架：JBoss的Marshalling包，google的Protobuf，基于Protobuf的Kyro（dubbo），MessagePack框架

2 前置工作 2.1 NettyUtil工具类

@Slf4j
public class NettyUtil {

    private static ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

    static {
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("jms-thread-pool");
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
    }

    public static void initServer(int port, Supplier> supplier) {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        List handlers = supplier.get();
                        for (ChannelHandler handler: handlers) {
                            socketChannel.pipeline().addLast(handler);
                        }
                    }
                });
        try {
            // 绑定端口，同步等待成功
            ChannelFuture future = serverBootstrap.bind(port).sync();
            System.out.println("netty server start at port:" + port);
            // 等待服务器监听端口关闭
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    public static void initClient(int port, Supplier> supplier) {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .handler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        List handlers = supplier.get();
                        for (ChannelHandler handler: handlers) {
                            socketChannel.pipeline().addLast(handler);
                        }
                    }
                });
        try {
            // 绑定端口，同步等待成功
            ChannelFuture future = bootstrap.connect(new InetSocketAddress(port)).sync();
            // 等待服务器监听端口关闭
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            group.shutdownGracefully();
        }finally {
            executor.execute(() -> {
                try {
                    Thread.sleep(1000);
                    log.info("server closed, start reconnect");
                    initClient(port, supplier);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }

}

2.2 传输实体类

@NoArgsConstructor
@AllArgsConstructor
//@Message
@Data
public class UserInfo implements Serializable {

    private Integer id;

    private String name;

    private Integer age;

    private BigDecimal salary;

}

3 Marshalling编解码

JBoss的Marshalling序列化框架，它是JBoss内部使用的序列化框架，Netty提供了Marshalling编码和解码器，方便用户在Netty中使用Marshalling。
JBoss Marshalling是一个Java对象序列化包，对JDK默认的序列化框架进行了优化，但又保持跟java.io.Serializable接口的兼容，同时增加了一些可调的参数和附加的特性，这些参数和特性可通过工厂类进行配置。

3.1 maven依赖

		
			org.jboss.marshalling
			jboss-marshalling
			2.0.5.Final
		

		
			org.jboss.marshalling
			jboss-marshalling-serial
			2.0.5.Final

3.2 Marshalling编解码器工厂

public class MarshallingCodeFactory {

    
    public static MarshallingDecoder buildMarshallingDecoder() {
        // 首先通过Marshalling工具类的静态方法获取MarshallerFactory实例对象 参数serial标识创建的是java序列化工厂对象。
        final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
        // 创建了MarshallingConfiguration对象，配置了版本号为5
        final MarshallingConfiguration configuration = new MarshallingConfiguration();
        configuration.setVersion(5);
        // 根据marshallerFactory和configuration创建provider
        UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
        // 构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
        return new MarshallingDecoder(provider, 1024);
    }

    
    public static MarshallingEncoder buildMarshallingEncoder() {
        final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
        final MarshallingConfiguration configuration = new MarshallingConfiguration();
        configuration.setVersion(5);
        MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
        // 构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
        return new MarshallingEncoder(provider);
    }
}

3.3 服务端

public class NettyServer {

    public static void main(String[] args) {

        NettyUtil.initServer(8080, () -> {
            List handlers = new ArrayList<>();
            handlers.add(MarshallingCodeFactory.buildMarshallingDecoder());
            handlers.add(MarshallingCodeFactory.buildMarshallingEncoder());
            handlers.add(new NettyServerHandler());
            return handlers;
        });
    }

}

class NettyServerHandler extends ChannelInboundHandlerAdapter {

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

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server receive msg:" + msg.getClass() + ":" + msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 出于性能考虑，为了防止select频繁唤醒Selector发送消息，netty的write方法不直接把消息写入
        // SocketChannel中，只是把消息放入缓冲数组，再调用flush方法写入SocketChannel
        ctx.flush();
    }
}

3.4 客户端

public class NettyClient {

    public static void main(String[] args) {

        NettyUtil.initClient(8080, () -> {
            List handlers = new ArrayList<>();
            handlers.add(MarshallingCodeFactory.buildMarshallingDecoder());
            handlers.add(MarshallingCodeFactory.buildMarshallingEncoder());
            handlers.add(new NettyClientHandler());
            return handlers;
        });
    }

}

class NettyClientHandler extends ChannelInboundHandlerAdapter {

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

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for(int i = 0; i < 100; i ++) {
            ctx.write(new UserInfo(1, "小明", 12, new BigDecimal(1234)));
        }
        ctx.flush();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("client receive msg:" + msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 出于性能考虑，为了防止select频繁唤醒Selector发送消息，netty的write方法不直接把消息写入
        // SocketChannel中，只是把消息放入缓冲数组，再调用flush方法写入SocketChannel
        ctx.flush();
    }
}

4 MessagePack编解码

Messagepack是一个高效的二进制序列化框架，支持多语言，跟json很像，但是性能更快，序列化后的码流更小。

4.1 maven依赖

		
			org.msgpack
			msgpack
			0.6.12

4.2 编写编解码器

public class MessagePackDecoder extends MessageToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List

Netty学习二-编解码技术

Java相关栏目本月热门文章