Netty 编码解码器

2022-03-05 09:02:48

　　netty 发送或者接受一个消息的时候，将会发生一次数据转换。入站消息会被解码，从字节转换为另一种格式，比如Java 对象；如果是出站消息会被编码成字节码。

　　测试自己编写一个编码器和解码器。

1. MyLongToByteEncoder Long型转为byte的编码器：实际上继承了ChannelOutboundHandlerAdapter 这个，是出站写出消息时用的编码器

package netty.coder;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
        // 父类的write方法如下。acceptOutboundMessage 会根据msg 是否是应该处理的数据，如果是就进行encode 编码处理； 如果不是就跳过。 也就是写入的数据类型是Long 的情况下会被 Encode 编码
        /**
         *     @Override
         *     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
         *         ByteBuf buf = null;
         *         try {
         *             if (acceptOutboundMessage(msg)) {
         *                 @SuppressWarnings("unchecked")
         *                 I cast = (I) msg;
         *                 buf = allocateBuffer(ctx, cast, preferDirect);
         *                 try {
         *                     encode(ctx, cast, buf);
         *                 } finally {
         *                     ReferenceCountUtil.release(cast);
         *                 }
         *
         *                 if (buf.isReadable()) {
         *                     ctx.write(buf, promise);
         *                 } else {
         *                     buf.release();
         *                     ctx.write(Unpooled.EMPTY_BUFFER, promise);
         *                 }
         *                 buf = null;
         *             } else {
         *                 ctx.write(msg, promise);
         *             }
         *         } catch (EncoderException e) {
         *             throw e;
         *         } catch (Throwable e) {
         *             throw new EncoderException(e);
         *         } finally {
         *             if (buf != null) {
         *                 buf.release();
         *             }
         *         }
         *     }
         */
        System.out.println("MyLongToByteEncoder 被调用");
        System.out.println("msg: " + msg);
        out.writeLong(msg);
    }
}

2. MyByteToLongDecoder byte转为long的解码器

package netty.coder;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class MyByteToLongDecoder extends ByteToMessageDecoder {

    /**
     * 该方法会根据传入的字节数，调用多次；直到确定没有新的元素添加到list。如果list不为空，则会传递到下一个handler， 也就是下一个handler 也会被调用多次
     *
     * @param ctx
     * @param in
     * @param out
     * @throws Exception
     */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyByteToLongByteMessage 被调用~~~");
        // long 是8个字节
        if (in.readableBytes() >= 8) {
            out.add(in.readLong());
        }
    }
}

3. MyServerHandler 处理器

package netty.coder;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class MyServerHandler extends SimpleChannelInboundHandler<Long> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        System.out.println("从客户端 " + ctx.channel().remoteAddress() + " 读取到的消息， long: " + msg);

        // 回显一个消息给客户端
        ctx.channel().writeAndFlush(123456L);
    }

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

4. MyNettyServerInitializer 初始化器

package netty.coder;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

public class MyNettyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        // 向管道加入处理器
        // 得到管道
        ChannelPipeline pipeline = ch.pipeline();

        // 加入一个netty 提供的httpServerCodec codec =>[coder - decoder]
        // 1. MyByteToLongDecoder 入站的解码器
//        pipeline.addLast(new MyByteToLongDecoder());
        pipeline.addLast(new MyByteToLongDecoder2());

        // 2. 增加一个出站的编码器
        pipeline.addLast(new MyLongToByteEncoder());
        // 3. 增加一个自定义的handler
        pipeline.addLast(new MyServerHandler());

        System.out.println("server is ok~~~~");
    }
}

5. MyServer

package netty.coder;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class MyServer {

    private static final Integer PORT = 6666;

    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new MyNettyServerInitializer());

        ChannelFuture sync = serverBootstrap.bind(PORT).sync();
        sync.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println("服务端启动成功，监听地址: " + PORT);
                }
            }
        });
    }
}

6. MyClientHandler 客户端处理器

package netty.coder;


import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义服务器端处理handler，需要继承netty定义的 ChannelInboundHandlerAdapter 类
 */
public class MyClientHandler extends SimpleChannelInboundHandler<Long> {

    private static final Logger log = LoggerFactory.getLogger(MyClientHandler.class);

    /**
     * 通道就绪事件
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("MyClientHandler ctx: " + ctx);
        ctx.writeAndFlush(123456789L);
        // 写入一个字符串也会被转为Long 型处理。 16 个字节正好是两个8byte。发送字符串不会调用MyLongToByteEncoder 进行编码。
//        ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        log.info("服务器地址：" + ctx.channel().remoteAddress() + "; 服务器会送的消息是：msg: " + msg);
    }

    /**
     * 发生异常事件
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

7. MyNettyClient

package netty.coder;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class MyNettyClient {

    private static final Integer PORT = 6666;

    public static void main(String[] args) throws InterruptedException {
        // 创建一个事件循环组
        EventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            // 创建一个启动Bootstrap(注意是Netty包下的)
            Bootstrap bootstrap = new Bootstrap();
            // 链式设置参数
            bootstrap.group(eventExecutors) // 设置线程组
                    .channel(NioSocketChannel.class) // 设置通道class
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // 1. MyByteToLongDecoder 入站的解码器
//                            pipeline.addLast(new MyByteToLongDecoder());
                            pipeline.addLast(new MyByteToLongDecoder2());
                            // 2. 增加一个出站的编码器
                            pipeline.addLast(new MyLongToByteEncoder());
                            // 3. 加入一个自定义的处理器
                            pipeline.addLast(new MyClientHandler());
                        }
                    });
            System.out.println("客户端is ok...");

            // 启动客户端连接服务器(ChannelFuture 是netty的异步模型)
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", PORT).sync();
            // 监听关闭通道
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 关闭
            eventExecutors.shutdownGracefully();
        }
    }
}

　　这里需要注意如果客户端发的是字符串，服务器端够8个字节也会转成Long 处理。并且多个字节，服务器端处理的时候会多次调用解码器解码然后调用后续的自定义处理器进行处理。

8. ReplayingDecoder 还有另一种Decoder

　　该类无需调用in.readableBytes 进行判断。泛型S指定了用户状态管理的类型，其中Void 表示不需要状态管理。例如：

package netty.coder;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

/**
 * ReplayingDecoder 使用这个类，我们无需调用in.readableBytes 进行判断。泛型S指定了用户状态管理的类型，其中Void 表示不需要状态管理
 */
public class MyByteToLongDecoder2 extends ReplayingDecoder<Void> {


    /**
     * 该方法会根据传入的字节数，调用多次；直到确定没有新的元素添加到list。如果list不为空，则会传递到下一个handler， 也就是下一个handler 也会被调用多次
     *
     * @param ctx
     * @param in
     * @param out
     * @throws Exception
     */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyByteToLongDecoder2 extends ReplayingDecoder 被调用~~~");
        // 无需判断是否满足8个字节
        out.add(in.readLong());
    }
}

其也有局限性：

1. 并不是所有的ByteBuf 都被支持，如果调用了一个不被支持的方法，将会抛出一个异常

2. ReplayingDecoder 在某些情况下可能会慢于ByteToMessageDecoder，例如网络缓慢并且消息格式复杂，消息会被拆成多个碎片，速度变慢

其他一些解码器：

1. LineBasedFrameDecoder，这个类在netty 内部也有使用，它使用行尾控制字符(\n活\r\n) 作为分隔符来解析数据

2. DelimiterBasedFrameDecoder 使用自定义的椰树字符作为消息的分隔符

3. HttpObjectDecoder 一个Http 数据的解码器

4. LengthFieldBasedFrameDecoder 通过制定长度来表示整包消息，这样就可以自动的处理粘包和半包消息

Netty 编码解码器

码农公寓

相关文章