Netty最简实践

前言

Netty 的本质是一个网络应用程序框架，拥有高性能、高吞吐量、低延迟、ByteBuf 零复制等优点。对 JDK 自带的 NIO 进行封装， Netty 做的更多更好，支持常用应用层协议，解决传输沾包半包现象，支持流量整形等。

引入依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.74.Final</version>
</dependency>

服务端

解码器

public class MyFrameDecoder extends LengthFieldBasedFrameDecoder {
    public MyFrameDecoder() {
        super(10240, 0, 2, 0, 2);
    }
}

继承 Netty 包下 LengthFieldBasedFrameDecoder 根据指定长度解决 沾包半包问题

参数说明

第一个参数 10240 :最大长度 如果长度大于这个值会抛出 TooLongFrameException

第二个参数 0: 长度字段偏移量。

第三个参数 2: 长度字段的长度

例如 05Hello 05是长度字段(他的长度为2) Hello 刚好是5个长度 05Hello 代表一个完整的消息

第四个参数 0: 要添加到长度字段值的补偿值

第五个参数 2: 解码后的数据为 05Hello 跳过长度字段05 最终获取的值为 Hello

编码器

public class MyFrameEncoder extends LengthFieldPrepender {
    public MyFrameEncoder() {
      	//长度字段的长度与解码器对应
        super(2);
    }
}

服务端消息处理

public class MsgServerProcessHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
				//读取客户端发来的消息
        System.out.println("读取:" + msg);
        if (ctx.channel().isActive() && ctx.channel().isWritable()) {
          	//返回给客户端的值
            ctx.writeAndFlush("Hello Netty");
        } else {
            System.out.println("not writable now, message dropped");
        }
    }
}

继承自 Netty 包下 SimpleChannelInboundHandler 类，此类中帮我们自动释放了 ByteBuf。

服务端完整代码

public class Server {
    public static void main(String[] args) throws InterruptedException {

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
				//这里使用的是 非主从 Reactor 多线程模式
      	//他会根据 cpu 创建一个最优的线程数
        serverBootstrap.group(new NioEventLoopGroup());

        serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
								//使用上面写的编解码器 定义解析规则
                pipeline.addLast(new MyFrameDecoder());
                pipeline.addLast( new MyFrameEncoder());
								//这里为了简单使用 Netty 自带的处理字符串
                pipeline.addLast(new StringDecoder());
                pipeline.addLast(new StringEncoder());

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

        ChannelFuture channelFuture = serverBootstrap.bind(9090).sync();
        channelFuture.channel().closeFuture().sync();
    }
}

特别注意：ChannelPipeline 中的顺序是有约束的。从上往下是读取数据，分别是 MyFrameDecoder()，StringDecoder()；从下往上是写出数据 分别是 StringEncoder(), MyFrameEncoder()，处理逻辑 MsgServerProcessHandler 放在最后。

客户端

解码器

public class MyFrameDecoder extends LengthFieldBasedFrameDecoder {
    public MyFrameDecoder() {
        super(Integer.MAX_VALUE, 0, 2, 0, 2);
    }
}

参数说明参考服务端解码器

编码器

public class MyFrameEncoder extends LengthFieldPrepender {
    public MyFrameEncoder() {
        super(2);
    }
}

客户端消息处理

public class ResponseDispatcherHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("client read:" + msg);
    }
}

客户端完整代码

public class Client {

    public static void main(String[] args) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(new NioEventLoopGroup());

        LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);

        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();

                pipeline.addLast(new MyFrameDecoder());
                pipeline.addLast(new MyFrameEncoder());

                pipeline.addLast(new StringDecoder());
                pipeline.addLast(new StringEncoder());

                pipeline.addLast(new ResponseDispatcherHandler());
                pipeline.addLast(loggingHandler);
            }
        });

        ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9090);
        channelFuture.sync();

        channelFuture.channel().writeAndFlush("Hello World1");
        channelFuture.channel().writeAndFlush("Hello World2");

        channelFuture.channel().closeFuture().sync();
    }
}

运行

先运行服务端，再运行客户端。

服务端打印如下：

读取:Hello World1
读取:Hello World2

客户端打印如下:

client read:Hello Netty
client read:Hello Netty

总结

源码连接 https://github.com/pepsiyoung/hexo-blog-demo/tree/main/console/src/main/java/netty