devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【netty】三万字详解！JAVA高性能通信框架，关于netty，看这一篇就够了_java 接收设备上送的不同协议的报文并解析 netty

作者：知新_RL | 2024-07-12 05:53:55

赞

踩

java 接收设备上送的不同协议的报文并解析 netty

目录

1.概述

2.hello world

3.EventLoop

4.channel

4.1.同步

4.2.异步

4.3.调试

4.4.关闭

4.5.为什么要用异步

5.future

6.promise

7.pipeline

8.byteBuf

8.1.创建

8.2.内存模式和池化

8.2.1.内存模式

8.2.2.池化

8.3.组成

8.4.操作

8.4.1.读写

8.4.2.释放

8.5.零拷贝

8.5.1.slice

8.5.2.composite

8.6.工具类

9.双向通信

10.粘包半包

10.1.问题成因

10.2.解决办法

10.2.1.短连接

10.2.2.解码器

1.概述

2.定长解码器

3.行解码器

4.固定帧长的解码器

11.协议解析

11.1.Redis

11.2.Http

12.协议设计

12.1.概述

12.2.编码

1.概述

netty，说人话就是封装NIO做出来的一个JAVA高性能通信框架。在JAVA领域，有高性能网络通信需求的时候，绝大多数都会选择netty作为通信框架。

关于JAVA的通信，我猜想可能博主的另外两篇关于BIO和NIO的文章作为本文的导读会不错：

详解TCP-CSDN博客

详解JAVA Socket-CSDN博客

JAVA BIO_java的bio有哪些-CSDN博客

全网最清晰JAVA NIO，看一遍就会-CSDN博客

netty底层就是封装的NIO。如果自己使用NIO的话至少会有以下的不便：

  • 需要自己构建协议。

  • 需要自己解决TCP传输问题，如粘包、半包。

  • API过于底层，不便于使用。

netty其实就是封装了一下NIO，使得NIO更便于使用。

2.hello world

依赖：

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

服务器：

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.channel.ChannelInboundHandlerAdapter;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.nio.NioServerSocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.string.StringDecoder;
  9. ​
  10. public class HelloServer {
  11.     public static void main(String[] args) {
  12.         //ServerBootstrap，启动器，负责组装netty组件
  13.         new ServerBootstrap()
  14.                 //1.怎样去接收IO？
  15.                 //事件组，事件组里面包含thread和selector，可以理解为netty种用来选择IO的组件
  16.                 .group(new NioEventLoopGroup())
  17.                 //2.接收成什么？
  18.                 //服务器ServerSocketChannel实现，由于上面用的Nio的事件组，所选nio的
  19.                 //除此以外，还支持BIO和特定操作系统的，如Linux的EpollServerSocketChannel
  20.                 .channel(NioServerSocketChannel.class)
  21.                 //3.做什么处理？
  22.                 //支持用责任链模式来对收到的IO进行链式处理
  23.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  24.                     //连接建立后才会调用初始化方法
  25.                     @Override
  26.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  27.                         //指定解码方式
  28.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  29.                         //ChannelInboundHandlerAdapter接口是netty种让用户自定义handler的接口
  30.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  31.                             @Override
  32.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  33.                                 System.out.println(msg);
  34.                             }
  35.                         });
  36.                     }
  37.                 })
  38.                 //4.绑定监听端口
  39.                 .bind(8080);
  40.     }
  41. }

客户端：

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.ChannelInitializer;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.channel.socket.nio.NioSocketChannel;
  5. import io.netty.handler.codec.string.StringEncoder;
  6. ​
  7. import java.net.InetSocketAddress;
  8. ​
  9. public class HelloCleint {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         new Bootstrap()
  12.                 .group(new NioEventLoopGroup())
  13.                 //用什么进行发送？
  14.                 //可以是BIO，也可以是NIO，也可以是epoll
  15.                 .channel(NioSocketChannel.class)
  16.                 //处理器
  17.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  18.                     @Override
  19.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  20.                         //指定编码方式
  21.                         nioSocketChannel.pipeline().addLast(new StringEncoder());
  22.                     }
  23.                 })
  24.                 //连接到服务器
  25.                 .connect(new InetSocketAddress("localhost",8080))
  26.                 //同步通信
  27.                 .sync()
  28.                 //代表连接对象
  29.                 .channel()
  30.                 //发送数据
  31.                 .writeAndFlush("hello world");
  32.     }
  33. }

3.EventLoop

eventLoop，事件循环对象，是一个单线程执行器，本质上就是一条线程+一个selector，用来单线程监听处理IO事件。

实际使用上很少直接使用EventLoop，而是使用EventLoopGroup，EventLoopGroup的构造方法中可以指定其中的EventLoop数量。

eventLoop除了继承Netty体系类的一些标准化接口外，还继承了JDK中的ScheduledExecutorService，使得其自身具备线程池一切的能力。既然是线程池，就可以用来执行任务。

eventLoop执行普通任务：

  1. EventLoopGroup group =new NioEventLoopGroup(5);
  2. ​
  3.         group.next().submit(()->{
  4.             try {
  5.                 Thread.sleep(10000);
  6.                 System.out.println("success!");
  7.             } catch (Exception e) {
  8.                 e.printStackTrace();
  9.             }
  10.         });

eventLoop执行IO任务：

一个EventGroupLoop其实就是一条线程，用来处理一条通信连接。

  1. public static void main(String[] args) {
  2.         //ServerBootstrap，启动器，负责组装netty组件
  3.         new ServerBootstrap()
  4.                 //1.怎样去接收IO？
  5.                 //事件组，事件组里面包含thread和selector，可以理解为netty种用来选择IO的组件
  6.                 .group(new NioEventLoopGroup())
  7.                 //2.接收成什么？
  8.                 //服务器ServerSocketChannel实现，由于上面用的Nio的事件组，所选nio的
  9.                 //除此以外，还支持BIO和特定操作系统的，如Linux的EpollServerSocketChannel
  10.                 .channel(NioServerSocketChannel.class)
  11.                 //3.做什么处理？
  12.                 //支持用责任链模式来对收到的IO进行链式处理
  13.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  14.                     //连接建立后才会调用初始化方法
  15.                     @Override
  16.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  17.                         //指定解码方式
  18.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  19.                         //ChannelInboundHandlerAdapter接口是netty种让用户自定义handler的接口
  20.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  21.                             @Override
  22.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  23.                                 System.out.println(msg);
  24.                             }
  25.                         });
  26.                     }
  27.                 })
  28.                 //4.绑定监听端口
  29.                 .bind(8080);
  30.     }

netty还给了一种更加细粒度的分层，就是让一部分EventLoop来选择IO，一部分EventLoop来处理IO，说白了就是一部分EventLoop出selector，一部分EventLoop出Thread。

  1. public static void main(String[] args) {
  2.         //ServerBootstrap，启动器，负责组装netty组件
  3.         new ServerBootstrap()
  4.                 //boss线程只负责accept事件，worker线程只负责io读写
  5.                 .group(new NioEventLoopGroup(),new NioEventLoopGroup())
  6.                 .channel(NioServerSocketChannel.class)
  7.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  8.                     @Override
  9.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  10.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  11.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  12.                             @Override
  13.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  14.                                 System.out.println(msg);
  15.                             }
  16.                         });
  17.                     }
  18.                 })
  19.                 .bind(8080);
  20.     }

4.channel

channel,对NIO的channel的二次封装，内核段缓冲区的抽象。不管是服务端还是客户端，只要调用channel()方法都能获取当前工作的这条channel。channel无非要注意的点就是它的同步和异步。

在实际应用中我们要知道在读的时候同步和异步是没有意义的，不可能在读IO的时候还区分同步读或者异步读，只可能是准备好了就读。只有写IO的时候区分同步和异步才是意义。所以在netty体系里很少会去服务端操作channel的同步和异步，一般都是在客户端操作channel的同步和异步。

4.1.同步

服务端：

在服务端让建立连接的时候休眠3秒。

  1. public static void main(String[] args) {
  2.         //ServerBootstrap，启动器，负责组装netty组件
  3.         new ServerBootstrap()
  4.                 //1.怎样去接收IO？
  5.                 //事件组，事件组里面包含thread和selector，可以理解为netty种用来选择IO的组件
  6.                 .group(new NioEventLoopGroup())
  7.                 //2.接收成什么？
  8.                 //服务器ServerSocketChannel实现，由于上面用的Nio的事件组，所选nio的
  9.                 //除此以外，还支持BIO和特定操作系统的，如Linux的EpollServerSocketChannel
  10.                 .channel(NioServerSocketChannel.class)
  11.                 //3.做什么处理？
  12.                 //支持用责任链模式来对收到的IO进行链式处理
  13.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  14.                     //连接建立后才会调用初始化方法
  15.                     @Override
  16.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  17.                         //指定解码方式
  18.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  19.                         //ChannelInboundHandlerAdapter接口是netty种让用户自定义handler的接口
  20.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  21.                             @Override
  22.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  23.                                 Thread.sleep(3000);
  24.                                 System.out.println(msg);
  25.                             }
  26.                         });
  27.                     }
  28.                 })
  29.                 //4.绑定监听端口
  30.                 .bind(8080);
  31.     }

客户端：

客户端使用channel的sync来进行同步通信，同步模式下在connect建立连接的时候，主线程会同步等待，连接建立后再向下执行。

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.ChannelInitializer;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.channel.socket.nio.NioSocketChannel;
  5. import io.netty.handler.codec.string.StringEncoder;
  6. ​
  7. import java.net.InetSocketAddress;
  8. ​
  9. public class HelloCleint {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         new Bootstrap()
  12.                 .group(new NioEventLoopGroup())
  13.                 //用什么进行发送？
  14.                 //可以是BIO，也可以是NIO，也可以是epoll
  15.                 .channel(NioSocketChannel.class)
  16.                 //处理器
  17.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  18.                     @Override
  19.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  20.                         //指定编码方式
  21.                         nioSocketChannel.pipeline().addLast(new StringEncoder());
  22.                     }
  23.                 })
  24.                 //连接到服务器
  25.                 .connect(new InetSocketAddress("localhost",8080))
  26.                 //同步通信
  27.                 .sync()
  28.                 //代表连接对象
  29.                 .channel()
  30.                 //发送数据
  31.                 .writeAndFlush("hello world");
  32.     }
  33. }

4.2.异步

channel默认处于异步通信模式，connect建立连接的时候，不会同步等待，而是会继续向下执行，由于服务器端延迟了3秒来建立连接，所以客户端发送这条“hello server”发送时，连接并未建立完成，最终效果就是丢包，服务器收不到这条数据。

  1. public static void main(String[] args) throws InterruptedException {
  2.         ChannelFuture channelFuture = new Bootstrap()
  3.                 .group(new NioEventLoopGroup())
  4.                 //用什么进行发送？
  5.                 //可以是BIO，也可以是NIO，也可以是epoll
  6.                 .channel(NioSocketChannel.class)
  7.                 //处理器
  8.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  9.                     @Override
  10.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  11.                         //指定编码方式
  12.                         nioSocketChannel.pipeline().addLast(new StringEncoder());
  13.                     }
  14.                 })
  15.                 //连接到服务器
  16.                 .connect(new InetSocketAddress("localhost", 8080));
  17.         //异步
  18.         channelFuture.channel().writeAndFlush("hello world");
  19.     }

当然，在异步通信上，netty支持了监听器，建立连接完成后，用事件回调的方式触发监听器。利用监听器，可以使得异步通信不丢包：

  1. //异步
  2. channelFuture.addListener(new ChannelFutureListener() {
  3.   @Override
  4.   public void operationComplete(ChannelFuture channelFuture) throws Exception {
  5.     channelFuture.channel().writeAndFlush("hello world");
  6.   }
  7. });

用监听器发送数据后，在当前业务场景下，即使服务端延迟了三秒才建立连接，但是任然能收到“hello world”这条消息。

4.3.调试

EmbeddedChannel是Netty中提供的一种特殊类型的Channel实现，主要用于单元测试。它允许你在测试中模拟输入事件（例如读取数据、写入数据）并检查输出事件（例如读取到的数据）。使用EmbeddedChannel可以在不启动真实的网络连接的情况下测试你的ChannelHandler逻辑。

代码示例：

自定义一个handler:

  1. import io.netty.channel.ChannelHandlerContext;
  2. import io.netty.channel.SimpleChannelInboundHandler;
  3. ​
  4. public class UpperCaseHandler extends SimpleChannelInboundHandler<String> {
  5.     @Override
  6.     protected void channelRead0(ChannelHandlerContext ctx, String msg) {
  7.         String upperCaseMsg = msg.toUpperCase();
  8.         ctx.writeAndFlush(upperCaseMsg);
  9.     }
  10. }

测试:

  1. import io.netty.channel.embedded.EmbeddedChannel;
  2. import org.junit.jupiter.api.Test;
  3. ​
  4. import static org.junit.jupiter.api.Assertions.assertEquals;
  5. ​
  6. public class UpperCaseHandlerTest {
  7.     @Test
  8.     public void testUpperCaseHandler() {
  9.         // 创建EmbeddedChannel，并添加要测试的Handler
  10.         EmbeddedChannel channel = new EmbeddedChannel(new UpperCaseHandler());
  11. ​
  12.         // 写入一个字符串消息到Channel
  13.         channel.writeInbound("hello");
  14. ​
  15.         // 读取Channel的输出
  16.         String output = channel.readOutbound();
  17. ​
  18.         // 验证处理后的消息是否符合预期
  19.         assertEquals("HELLO", output);
  20. ​
  21.         // 关闭Channel
  22.         channel.finish();
  23.     }
  24. }

4.4.关闭

由于channel的close方法是异步的，所以在关闭资源时会存在风险。比如代码顺序为：

  • close掉channel

  • close掉其它资源

有可能在close掉其它资源的时候，channel并没有close掉，也就可能出现，channel中还有数据没处理完，其它资源被关掉了，导致数据处理失败的问题。所以更为稳妥的方式是用同步的机制来关闭channel。netty中封装了CloseFuture来同步关闭channel。

  1. ChannelFuture closeFuture = channelFuture.channel().closeFuture();
  2. //同步关闭
  3. closeFuture.sync();

要注意的是channel停止后如果EventLoopGroup还有其它线程时，程序是不会中止的，想要中止程序，必须再close掉group，EventLoopGroup提供了优雅停机的API——shutdownGracefully，会先停止接收请求，驻留的请求处理完成后，关掉group。

4.5.为什么要用异步

我们可以看到channel里面大量的用到了异步，对一个channel的操作，connect是一条线程，write是一条线程，close也是一条线程......

用异步的方式来处理，不仅不会加快单个IO任务的速度，反而还会略微拉长一个IO的响应时间，但是异步能明显提高吞吐量。

举个例子，一个病人看病，分为挂号、看病、缴费。取药，同步的方式就是一个医生走完一个病人的所有流程：

而异步的方式就是医生分工合作，每个医生单独负责一个项目，这样一个时间段内虽然处理的任务综合是一样的，但是在峰值的吞吐量上，异步是同步的四倍：

5.future

JDK的future是表示一个任务，netty的future是对JDK的future做了二次封装。

同步：

  1. public static void main(String[] args) throws Exception {
  2.         NioEventLoopGroup nioEventLoopGroup=new NioEventLoopGroup();
  3.         Future<String> future = nioEventLoopGroup.submit(new Callable<String>() {
  4.             public String call() throws Exception {
  5.                 Thread.sleep(1000);
  6.                 return "success!";
  7.             }
  8.         });
  9.         //future的get方法是同步的，同步等待线程返回返回值为止
  10.         System.out.println(future.get());
  11.     }

异步：

用监听器实现异步

  1. public static void main(String[] args) throws Exception {
  2.         NioEventLoopGroup nioEventLoopGroup=new NioEventLoopGroup();
  3.         Future<String> future = nioEventLoopGroup.submit(new Callable<String>() {
  4.             public String call() throws Exception {
  5.                 Thread.sleep(1000);
  6.                 return "success!";
  7.             }
  8.         });
  9.         //用监听器来实现异步
  10.         future.addListener(new GenericFutureListener<Future<? super String>>() {
  11.             public void operationComplete(Future<? super String> future) throws Exception {
  12.                 System.out.println(future.get());
  13.             }
  14.         });
  15.     }

6.promise

光是有future是不够的，因为future必须处理完了，才能拿到结果，有些时候需要提前拿到结果开始处理，就需要在两个线程间进行通信，通信就需要一个存放数据的地方，也就有了promise，其可以理解为一个数据容器，可以向该容器中手动的存放数据、拿数据。

  1. import io.netty.channel.EventLoop;
  2. import io.netty.channel.EventLoopGroup;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.util.concurrent.DefaultPromise;
  5.  
  6. public static void main(String[] args) {
  7.         EventLoopGroup eventLoopGroup=new NioEventLoopGroup();
  8.         EventLoop eventLoop = eventLoopGroup.next();
  9.         final DefaultPromise<String> promise=new DefaultPromise<String>(eventLoop);
  10.         eventLoop.execute(new Runnable() {
  11.             public void run() {
  12.                 try {
  13.                     Thread.sleep(10000);
  14.                 } catch (InterruptedException e) {
  15.                     e.printStackTrace();
  16.                 }
  17.                 promise.setSuccess("success!");
  18.             }
  19.         });
  20.         //默认是同步
  21.         try {
  22.             System.out.println(promise.get());
  23.         } catch (Exception e) {
  24.             e.printStackTrace();
  25.         }
  26.         //可以用监听器来实现异步
  27.         //promise.addListener(new GenericFutureListener<Future<? super String>>() {
  28.             //public void operationComplete(Future<? super String> future) throws Exception {
  29.                 //System.out.println(promise.get());
  30.             //}
  31.         //});
  32.     }

promise支持向外抛异常：

  1. import io.netty.channel.EventLoop;
  2. import io.netty.channel.EventLoopGroup;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.util.concurrent.DefaultPromise;
  5.  
  6. public class PromiseDemo {
  7.     public static void main(String[] args) {
  8.         EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  9.         EventLoop eventLoop = eventLoopGroup.next();
  10.         final DefaultPromise<String> promise = new DefaultPromise<String>(eventLoop);
  11.         eventLoop.execute(new Runnable() {
  12.             public void run() {
  13.                 try {
  14.                     int i = 1 / 0;
  15.                 } catch (Exception e) {
  16.                     promise.setFailure(e);
  17.                 }
  18.             }
  19.         });
  20.  
  21.         try {
  22.             System.out.println(promise.get());
  23.         } catch (Exception e) {
  24.             e.printStackTrace();
  25.         }
  26.     }
  27. }

7.pipeline

netty中使用了责任链来处理对channel的读写请求，链上每一个节点都是一个处理器，有两种处理器：

  • 出站处理器，用来处理write操作。

  • 入站处理器，用来处理read操作。

这里要注意，是出战、入站，不是出栈、入栈。Netty 的设计参考了这种网络协议栈的思想，所以出站（Outbound）和入站（Inbound）这两个概念是遵循网络协议栈的传统命名。

  • 出站（Outbound）：数据从应用程序流向网络的过程被称为“出站”，因为数据是从应用程序向外发送，穿越协议栈的各个层级，最终到达网络。

    在 Netty 中，ChannelOutboundHandlerAdapter 处理的是数据从应用程序到网络的过程，即数据从上层（应用层或业务层）向下层（传输层、网络层、数据链路层等）传递的过程。

  • 入站（Inbound）：数据从网络流向应用程序的过程被称为“入站”，因为数据是从外部网络进入应用程序，穿越协议栈的各个层级，最终到达应用程序。

    在 Netty 中，ChannelInboundHandlerAdapter 处理的是数据从网络到应用程序的过程，即数据从下层（传输层、网络层、数据链路层等）向上层（应用层或业务层）传递的过程。

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.*;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.channel.socket.nio.NioServerSocketChannel;
  5. import io.netty.channel.socket.nio.NioSocketChannel;
  6.  
  7. public class Server {
  8.     public static void main(String[] args) {
  9.         new ServerBootstrap()
  10.                 .group(new NioEventLoopGroup())
  11.                 .channel(NioServerSocketChannel.class)
  12.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  13.                     @Override
  14.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  15.                         //H1->H2->H3->h4->h5->h6
  16.                         //入站处理器
  17.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  18.                             @Override
  19.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 							{
  20.                                 System.out.println("H1");
  21.                                 //向下走
  22.                                 super.channelRead(ctx,msg);
  23.                             }
  24.                         });
  25.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  26.                             @Override
  27.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 							{
  28.                                 System.out.println("H2");
  29.                                 //向下走
  30.                                 super.channelRead(ctx,msg);
  31.                             }
  32.                         });
  33.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  34.                             @Override
  35.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 							{
  36.                                 System.out.println("H3");
  37.                                 //写操作，用来触发后面的出站处理器
  38.                                 	 	nioSocketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Servers......".getBytes()));
  39.                             }
  40.                         });
  41.                         //出站处理器
  42.                         nioSocketChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
  43.                             @Override
  44.                             public void write(ChannelHandlerContext ctx, Object o, ChannelPromise 		channelPromise) throws Exception {
  45.                                 System.out.println("h4");
  46.                                 super.write(ctx,o,channelPromise);
  47.                             }
  48.                         });
  49.                         nioSocketChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
  50.                             @Override
  51.                             public void write(ChannelHandlerContext ctx, Object o, ChannelPromise channelPromise) throws Exception {
  52.                                 System.out.println("h5");
  53.                                 super.write(ctx,o,channelPromise);
  54.                             }
  55.                         });
  56.                         nioSocketChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
  57.                             @Override
  58.                             public void write(ChannelHandlerContext ctx, Object o, ChannelPromise channelPromise) throws Exception {
  59.                                 System.out.println("h6");
  60.                                 super.write(ctx,o,channelPromise);
  61.                             }
  62.                         });
  63.                     }
  64.                 }).bind(8080);
  65.     }
  66. }

client:

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.ChannelInitializer;
  3. import io.netty.channel.nio.NioEventLoopGroup;
  4. import io.netty.channel.socket.nio.NioSocketChannel;
  5. import io.netty.handler.codec.string.StringEncoder;
  6.  
  7. import java.net.InetSocketAddress;
  8.  
  9. public class Client {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         new Bootstrap()
  12.                 .group(new NioEventLoopGroup())
  13.                 .channel(NioSocketChannel.class)
  14.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  15.                     @Override
  16.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  17.                         nioSocketChannel.pipeline().addLast(new StringEncoder());
  18.                     }
  19.                 })
  20.                 .connect(new InetSocketAddress("localhost", 8080))
  21.                 .sync()
  22.                 .channel()
  23.                 .writeAndFlush("hello world");
  24.     }
  25. }

server端的输出结果：

入站处理器顺序执行，出栈处理器逆序执行。

8.byteBuf

8.1.创建

在Java NIO（New I/O）中，ByteBuffer 是一个用来处理字节数据的缓冲区类，是对NIO的byteBuffer的二次封装和扩展，可以直接理解为用户段内存的抽象。

开辟byteBuf：

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3.  
  4. public class test {
  5.     public static void main(String[] args) {
  6.     	//可以通过传参来指定大小
  7.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  8.     }
  9. }

8.2.内存模式和池化

8.2.1.内存模式

根据所开辟的内存空间的位置的不同，byteBuf分为两类：

  • 直接缓冲区

  • 非直接缓冲区

直接缓冲区：

直接创建在物理机的缓冲区中，创建和销毁的代价昂贵，但是读写性能高。要注意的是直接内存不受GC的管理，需要注意手动释放内存，避免内存泄露。

创建池化的直接缓冲区：

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3.  
  4. public class test {
  5.     public static void main(String[] args) {
  6.         ByteBuf directBuffer = ByteBufAllocator.DEFAULT.directBuffer();
  7.     }
  8. }

非直接缓冲区：

创建在JVM中的缓冲区，创建和销毁的代价相对没那么高，但是读写性能相对较低。

创建池化的非直接缓冲区:

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3.  
  4. public class test {
  5.     public static void main(String[] args) {
  6.         ByteBuf directBuffer = ByteBufAllocator.DEFAULT.heapBuffer();
  7.     }
  8. }

8.2.2.池化

ByteBuf 的池化是指将 ByteBuf 实例预先分配并存储在内存池中，以便在需要时进行重复使用。池化 ByteBuf 的主要目的是减少内存分配和垃圾回收的开销，从而提高性能。Netty 提供了池化 ByteBuf 的功能，它内置了两种 ByteBuf 池化的实现：PooledByteBufAllocator 和 UnpooledByteBufAllocator。

1.PooledByteBufAllocator（池化的内存分配器）：

PooledByteBufAllocator 是 Netty 提供的默认的 ByteBuf 池化实现。它通过预先分配一些 ByteBuf 实例，并将它们存储在池中。当需要创建新的 ByteBuf 实例时，它会从池中获取已有的实例，而不是每次都重新分配内存。

使用 PooledByteBufAllocator 可以减少频繁的内存分配和释放操作，避免了堆内存的碎片化，提高了性能。

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3. import io.netty.buffer.PooledByteBufAllocator;
  4.  
  5. public class PooledByteBufExample {
  6.  
  7.     public static void main(String[] args) {
  8.         // 使用 PooledByteBufAllocator 创建 ByteBuf
  9.         ByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;
  10.         ByteBuf pooledBuffer = allocator.buffer(1024); // 创建1KB的池化 ByteBuf
  11.  
  12.         // 使用 pooledBuffer...
  13.  
  14.         // 释放 ByteBuf，将其返回到池中
  15.         pooledBuffer.release();
  16.     }
  17. }

2.UnpooledByteBufAllocator（非池化的内存分配器）：

UnpooledByteBufAllocator 是 Netty 提供的非池化的 ByteBuf 实现。它每次都会分配新的内存，不会重用已有的 ByteBuf 实例。虽然不会涉及到池的管理，但在一些短期存活或者需要手动管理内存的场景下使用非池化内存分配器可能更合适。

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3. import io.netty.buffer.UnpooledByteBufAllocator;
  4.  
  5. public class UnpooledByteBufExample {
  6.  
  7.     public static void main(String[] args) {
  8.         // 使用 UnpooledByteBufAllocator 创建 ByteBuf
  9.         ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
  10.         ByteBuf unpooledBuffer = allocator.buffer(1024); // 创建1KB的非池化 ByteBuf
  11.  
  12.         // 使用 unpooledBuffer...
  13.  
  14.         // 释放 ByteBuf（注意：在非池化情况下，需要手动释放 ByteBuf）
  15.         unpooledBuffer.release();
  16.     }
  17. }

在使用池化 ByteBuf 时，需要注意在不再使用 ByteBuf 时调用 release() 方法，将它返回到池中，以便被重用。这样可以避免内存泄漏和提高性能。

8.3.组成

10.Nettyä¹ByteBufä»ç»ï¼äºï¼01.png

bytebuf一开始有个初始化容量（capacity），可以手动指定，没有手动指定时也有个默认值。

bytebuf是自动扩容的，扩容的上限（max capacity）其实就是机器的物理内存。

读写指针一开始在0位，随着读写，读写指针向后移动。要注意，bytebuf的读写，只涉及指针的移动，不涉及内存的回收，也就是读过的区域（废弃字节）并不会被释放，除非调用特殊的API（discardReadBytes()）。

netty的bytebuf相较于NIO的bytebuffer，有以下优势：

  • bytebuffer读写公用一个指针，所以，读之前要切换到读模式；写之前要切换到写模式。

  • bytebuf自动扩容，而bytebuffer不行。

8.4.操作

8.4.1.读写

写操作:

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3. import java.nio.charset.StandardCharsets;
  4.  
  5. public class test {
  6.     public static void main(String[] args) {
  7.         //写入数字
  8.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  9.         buffer.writeInt(666);
  10.  
  11.         // 写入字符串
  12.         String stringValue = "Hello, World!";
  13.         byte[] stringBytes = stringValue.getBytes(StandardCharsets.UTF_8);
  14.         buffer.writeBytes(stringBytes);
  15.     }
  16. }

读操作:

  1. // 读取整数
  2. int readIntValue = buffer.readInt();
  3.  
  4. // 读取字符串
  5. int readableBytes = buffer.readableBytes();
  6. byte[] stringBytes = new byte[readableBytes];
  7. buffer.readBytes(stringBytes);
  8. String readStringValue = new String(stringBytes, StandardCharsets.UTF_8);

需要注意的是，在读取数据之前，你需要确保 ByteBuf 中有足够的可读字节数。可以使用 readableBytes() 方法来检查 ByteBuf 中的可读字节数。

此外，ByteBuf 还提供了其他的读写操作，比如 readableBytes() 用于获取可读字节数，writerIndex() 和 readerIndex() 用于获取写入和读取的索引位置等。在使用 ByteBuf 时，请确保在读写时不越界，并且注意释放 ByteBuf 以避免内存泄漏。在Netty中，通常会使用 ReferenceCountUtil.release(buffer) 来释放 ByteBuf，确保资源得到正确释放。

读写指针：

8.4.2.释放

bytebuf要特别注意资源的释放，以避免内存泄漏。Netty使用引用计数（Reference Counting）来管理 ByteBuf 的生命周期，确保在不再需要使用时及时释放资源。

在Netty中，release() 和 retain() 是用于管理 ByteBuf 引用计数的方法。

release() 方法用于将 ByteBuf 的引用计数减少1。当引用计数减至0时，Netty会释放 ByteBuf 的内存（如果使用了池化的 ByteBuf，则将它归还给池）。

ByteBuf buffer = //... 从某个地方获取ByteBuf实例
buffer.release(); // 引用计数减少1，如果引用计数为0，释放ByteBuf的内存

retain() 方法用于将 ByteBuf 的引用计数增加1。当你调用 retain() 方法时，你告诉Netty你对这个 ByteBuf 感兴趣，即使在你使用完后，其他代码也可能继续使用它。

ByteBuf buffer = //... 从某个地方获取ByteBuf实例
buffer.retain(); // 引用计数增加1，防止在使用完后被提前释放

8.5.零拷贝

零拷贝其实没有严格的定义，指的是减少IO过程中数据在内存中拷贝的次数这样一个大致目标。在netty的ByteBuf中也存在一些零拷贝机制，用来在多个ByteBuf之间进行数据传递。

8.5.1.slice

在 Netty 中，ByteBuf 的 slice() 方法用于创建一个与原始 ByteBuf 共享数据的新 ByteBuf。换句话说，slice() 方法返回一个从原始 ByteBuf 中截取出来的视图，这个视图与原始 ByteBuf 共享底层数据，但拥有自己的独立读写指针。由于是直接通过读写指针指向同一块内存的，所以slice出来的bytebuf并没有发送数据拷贝，是0拷贝。

如何理解拥有自己的独立读写指针喃？因为slice出来的buf和元buf共享内存，为了避免slice出来的buf通过写指针来进行写，进而影响元buf，netty在设计时故意就禁止了slice动用写指针来向元buf中进行写。只能通过读指针来读。

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  3.  
  4. public class test {
  5.     public static void main(String[] args) {
  6.     	//开一个容量为10字节的ByteBuf
  7.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
  8.         //写入数据
  9.         buffer.writeBytes(new byte[]{1,2});
  10.         //slice
  11.         ByteBuf slice = buffer.slice(0,2);
  12.         //因为slice不与元buf共享读写指针，所以write会报错，因为write是用的读写指针来进行读写，但是set不会报错，因为set不是用的读写指针来进行读写的。
  13.         //slice.writeByte(1);
  14.         slice.setByte(0,2);
  15.         while(slice.isReadable()){
  16.             System.out.println(slice.readByte());
  17.         }
  18.   
  19.     }
  20. }

8.5.2.composite

slice是将一个大的bytebuf划分成多个小的bytebuff，composite是将多个小的bytebuf聚合成一个大的bytebuf。

在 Netty 中，CompositeByteBuf 是 ByteBuf 的一个特殊实现，它提供了一种能够组合多个 ByteBuf 实例的方式。CompositeByteBuf 允许将多个 ByteBuf 视为一个单一的逻辑缓冲区，而不需要将它们合并成一个实际的连续内存块。这种设计可以提高内存的利用率和降低内存拷贝的次数。

  1. public static void main(String[] args) {
  2.         ByteBuf buffer1 = ByteBufAllocator.DEFAULT.buffer();
  3.         buffer1.writeBytes(new byte[]{1,2,3,4,5});
  4.         ByteBuf buffer2 = ByteBufAllocator.DEFAULT.buffer();
  5.         buffer2.writeBytes(new byte[]{6,7,8,9,10});
  6.         CompositeByteBuf compositeBuffer = ByteBufAllocator.DEFAULT.compositeBuffer();
  7.         //可变参数，可以有多个
  8.         compositeBuffer.addComponents(buffer1,buffer2);
  9.         while (compositeBuffer.isReadable()){
  10.             System.out.println(compositeBuffer.readByte());
  11.         }
  12.     }

8.6.工具类

Unpooled 是 Netty 提供的一个工具类，用于创建不需要池化的 ByteBuf 实例。在 Netty 中，ByteBuf 是用来操作字节数据的缓冲区类。通常，Unpooled 类提供了一些静态方法，用于创建不同类型的 ByteBuf 实例，包括堆缓冲区（heap buffer）、直接缓冲区（direct buffer）、组合缓冲区（composite buffer）等。

也就是说可以用unpooled来开辟各类型的bytebuf。

9.双向通信

服务端:

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9. import io.netty.handler.codec.string.StringDecoder;
  10.  
  11. public class Server {
  12.     public static void main(String[] args) {
  13.         //ServerBootstrap，启动器，负责组装netty组件
  14.         new ServerBootstrap()
  15.                 //1.怎样去接收IO？
  16.                 //事件组，事件组里面包含thread和selector，可以理解为netty种用来选择IO的组件
  17.                 .group(new NioEventLoopGroup())
  18.                 //2.接收成什么？
  19.                 //服务器ServerSocketChannel实现，由于上面用的Nio的事件组，所选nio的
  20.                 //除此以外，还支持BIO和特定操作系统的，如Linux的EpollServerSocketChannel
  21.                 .channel(NioServerSocketChannel.class)
  22.                 //3.做什么处理？
  23.                 //支持用责任链模式来对收到的IO进行链式处理
  24.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  25.                     //连接建立后才会调用初始化方法
  26.                     @Override
  27.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  28.                         //指定解码方式
  29.                         nioSocketChannel.pipeline().addLast(new StringDecoder());
  30.                         //ChannelInboundHandlerAdapter接口是netty种让用户自定义handler的接口
  31.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  32.                             @Override
  33.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  34.                                 System.out.println(msg);
  35.                                 ByteBuf response = ctx.alloc().buffer();
  36.                                 response.writeBytes(msg.toString().getBytes());
  37.                                 ctx.writeAndFlush(response);
  38.                             }
  39.                         });
  40.                     }
  41.                 })
  42.                 //4.绑定监听端口
  43.                 .bind(8080);
  44.     }
  45. }

客户端：

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.string.StringEncoder;
  9. import java.net.InetSocketAddress;
  10.  
  11. public class Client {
  12.     public static void main(String[] args) throws InterruptedException {
  13.         new Bootstrap()
  14.                 .group(new NioEventLoopGroup())
  15.                 //用什么进行发送？
  16.                 //可以是BIO，也可以是NIO，也可以是epoll
  17.                 .channel(NioSocketChannel.class)
  18.                 //处理器
  19.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  20.                     @Override
  21.                     protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  22.                         //指定编码方式
  23.                         nioSocketChannel.pipeline().addLast(new StringEncoder());
  24.                         nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  25.                             @Override
  26.                             public void channelActive(ChannelHandlerContext ctx) throws Exception {
  27.                                 ByteBuf buffer = ctx.alloc().buffer();
  28.                                 buffer.writeBytes("hello".getBytes());
  29.                                 ctx.writeAndFlush(buffer);
  30.                             }
  31.  
  32.                             @Override
  33.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  34.                                 System.out.println(msg.toString());
  35.                             }
  36.                         });
  37.                     }
  38.                 })
  39.                 //连接到服务器
  40.                 .connect(new InetSocketAddress("localhost", 8080));
  41.     }
  42. }

10.粘包半包

10.1.问题成因

粘包：发送abc def，接收到abcdef

半包：发送abcdef，接收到abc或者def

原因：

  • 没有清晰的结束符，导致不知道收到何处才是一个完成的包。

  • IO缓冲区大小过大或者过小，导致收太多或者收不完。

解决粘包和半包问题通常需要在设计通信协议时采取一些策略。

10.2.解决办法

10.2.1.短连接

解决粘包半包问题的其中一个办法是——短连接。

所谓短连接就是当一次完整的报文返送完成后，客户端主动断开TCP连接。粘包半包的根本原因其实就是不知道一个完整的报文何时收完，通过客户端发送完一次完整的信息后主动断开连接，让服务器端感知到，一次完整的信息发送完成。

客户端：

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.channel.ChannelFuture;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9.  
  10. public class Client {
  11.     public static void main(String[] args) throws InterruptedException {
  12.         for (int i=0;i<10;i++){
  13.             send();
  14.         }
  15.     }
  16.  
  17.     public static void send(){
  18.         NioEventLoopGroup worker = new NioEventLoopGroup();
  19.         try {
  20.             Bootstrap bootstrap = new Bootstrap();
  21.             bootstrap.channel(NioSocketChannel.class);
  22.             bootstrap.group(worker);
  23.             bootstrap.handler(
  24.                     new ChannelInitializer<NioSocketChannel>() {
  25.                         @Override
  26.                         protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
  27.                             //指定编码方式
  28.                             nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  29.                                 @Override
  30.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  31.                                     ByteBuf buffer = ctx.alloc().buffer();
  32.                                     buffer.writeBytes(new byte[]{0,1,2,3,4,5,6,7,8,9,10});
  33.                                     ctx.writeAndFlush(buffer);
  34.                                     ctx.channel().close();
  35.                                 }
  36.                             });
  37.                         }
  38.                     }
  39.             );
  40.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
  41.             channelFuture.channel().closeFuture().sync();
  42.         }catch(Exception e){
  43.             e.printStackTrace();
  44.         }finally {
  45.             worker.shutdownGracefully();
  46.         }
  47.     }
  48. }

服务器：

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9.  
  10. public class Server {
  11.     public static void main(String[] args) {
  12.         NioEventLoopGroup boss=new NioEventLoopGroup();
  13.         NioEventLoopGroup worker=new NioEventLoopGroup();
  14.         try{
  15.             ServerBootstrap serverBootstrap=new ServerBootstrap();
  16.             serverBootstrap.channel(NioServerSocketChannel.class);
  17.             serverBootstrap.group(boss,worker);
  18.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  19.                 @Override
  20.                 protected void initChannel(SocketChannel socketChannel) throws Exception {
  21.                     socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  22.                         @Override
  23.                         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  24.                             System.out.println(msg);
  25.                         }
  26.                     });
  27.                 }
  28.             });
  29.             ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
  30.             channelFuture.channel().closeFuture().sync();
  31.         }catch (Exception e){
  32.             e.printStackTrace();
  33.         }finally {
  34.             boss.shutdownGracefully();
  35.             worker.shutdownGracefully();
  36.         }
  37.     }
  38. }

10.2.2.解码器

1.概述

解码器是netty自带的一类用来从请求报文中解析出数据的handler。其底层原理都是从指定位置开始，解析出定长的字节内容来。

2.定长解码器

FixedLengthFrameDecoder，定长解码器，用来在报文中获取出指定长度的字节。

server：

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.buffer.ByteBuf;
  3. import io.netty.channel.*;
  4. import io.netty.channel.nio.NioEventLoopGroup;
  5. import io.netty.channel.socket.SocketChannel;
  6. import io.netty.channel.socket.nio.NioServerSocketChannel;
  7. import io.netty.handler.codec.FixedLengthFrameDecoder;
  8.  
  9. import java.nio.charset.StandardCharsets;
  10.  
  11. public class server {
  12.     public static void main(String[] args) throws InterruptedException {
  13.         ServerBootstrap serverBootstrap = new ServerBootstrap();
  14.         serverBootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
  15.                 .channel(NioServerSocketChannel.class)
  16.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  17.                     @Override
  18.                     protected void initChannel(SocketChannel ch) {
  19.                         ChannelPipeline pipeline = ch.pipeline();
  20.  
  21.                         // 定长解码器，每个消息长度固定为10个字节
  22.                         pipeline.addLast(new FixedLengthFrameDecoder(10));
  23.  
  24.                         // 业务处理器
  25.                         pipeline.addLast(new ChannelInboundHandlerAdapter() {
  26.                             @Override
  27.                             public void channelRead(ChannelHandlerContext ctx, Object msg) {
  28.                                 ByteBuf byteBuf = (ByteBuf) msg;
  29.                                 String content = byteBuf.toString(StandardCharsets.UTF_8);
  30.                                 System.out.println("Received message: " + content);
  31.                                 byteBuf.release(); // 释放ByteBuf资源
  32.                             }
  33.                         });
  34.                     }
  35.                 });
  36.  
  37.         Channel channel = serverBootstrap.bind(8080).sync().channel();
  38.         channel.closeFuture().sync();
  39.     }
  40. }
3.行解码器

Netty的行处理器（LineBasedFrameDecoder）是一种用于处理以换行符（\n）或回车换行符（\r\n）为消息分隔符的情况。它会按照换行符或回车换行符将接收到的数据切分成消息，适用于处理文本协议中每行代表一个消息的场景。

server：

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.channel.ChannelInboundHandlerAdapter;
  4. import io.netty.handler.codec.LineBasedFrameDecoder;
  5.  
  6. public class LineBasedServerHandler extends ChannelInboundHandlerAdapter {
  7.     public LineBasedServerHandler() {
  8.         // 添加行处理器到ChannelPipeline中，使用换行符作为消息分隔符
  9.         ctx.pipeline().addLast(new LineBasedFrameDecoder(1024));
  10.     }
  11.  
  12.     @Override
  13.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
  14.         // 这里的msg是一个ByteBuf，表示一个完整的行消息
  15.         ByteBuf buf = (ByteBuf) msg;
  16.         String line = buf.toString(io.netty.util.CharsetUtil.UTF_8);
  17.         System.out.println("Received message: " + line);
  18.         buf.release(); // 释放ByteBuf资源
  19.     }
  20.  
  21.     @Override
  22.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  23.         cause.printStackTrace();
  24.         ctx.close();
  25.     }
  26. }

客户端：

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.channel.ChannelInboundHandlerAdapter;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.ChannelPipeline;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9.  
  10. public class LineBasedClient {
  11.     public static void main(String[] args) throws InterruptedException {
  12.         Bootstrap bootstrap = new Bootstrap();
  13.         bootstrap.group(new NioEventLoopGroup())
  14.                 .channel(NioSocketChannel.class)
  15.                 .handler(new ChannelInitializer<SocketChannel>() {
  16.                     @Override
  17.                     protected void initChannel(SocketChannel ch) {
  18.                         ChannelPipeline pipeline = ch.pipeline();
  19.  
  20.                         // 添加行处理器到ChannelPipeline中，使用换行符作为消息分隔符
  21.                         pipeline.addLast(new LineBasedFrameDecoder(1024));
  22.                         
  23.                         // 客户端的业务处理器
  24.                         pipeline.addLast(new ChannelInboundHandlerAdapter() {
  25.                             @Override
  26.                             public void channelActive(ChannelHandlerContext ctx) {
  27.                                 // 发送带换行符的消息
  28.                                 String message = "Hello, Netty!\n";
  29.                                 ctx.writeAndFlush(message);
  30.                             }
  31.                         });
  32.                     }
  33.                 });
  34.  
  35.         bootstrap.connect("localhost", 8080).sync().channel().closeFuture().sync();
  36.     }
  37. }
4.固定帧长的解码器

Netty中的LengthFieldBasedFrameDecoder是一种用于解决粘包和半包问题的解码器。通信报文的结构说白了无非就是头部+身体，头部中记录关于消息长度等信息，身体中携带要传递的消息。LengthFieldBasedFrameDecoder就是根据设置的参数来准确的切分消息的头部和身体，就能确保每个消息被正确地接收和处理。

构造方法如下：

public LengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip)

  • maxFrameLength：指定消息的最大长度，超过这个长度的消息将被丢弃。

  • lengthFieldOffset：指定长度字段在消息中的偏移量。

  • lengthFieldLength：指定长度字段的长度，可以是1、2、3、4、8等字节。

  • lengthAdjustment：指定长度字段的值需要进行调整的偏移量，通常为消息头的长度。

  • initialBytesToStrip：指定解码时需要跳过的字节数，通常为长度字段的长度。

代码示例：

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.Unpooled;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  6.  
  7. public class LengthFieldServerHandler extends ChannelInboundHandlerAdapter {
  8.     public LengthFieldServerHandler() {
  9.         // 添加LengthFieldBasedFrameDecoder，指定各个参数
  10.         ctx.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
  11.     }
  12.  
  13.     @Override
  14.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
  15.         // 这里的msg是一个ByteBuf，表示一个完整的消息
  16.         ByteBuf buf = (ByteBuf) msg;
  17.         String message = buf.toString(io.netty.util.CharsetUtil.UTF_8);
  18.         System.out.println("Received message: " + message);
  19.         buf.release(); // 释放ByteBuf资源
  20.     }
  21.  
  22.     @Override
  23.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  24.         cause.printStackTrace();
  25.         ctx.close();
  26.     }
  27. }

11.协议解析

11.1.Redis

11.2.Http

名字里带codec的，在业内基本都是编解码器。

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.http.HttpContent;
  10. import io.netty.handler.codec.http.HttpRequest;
  11. import io.netty.handler.codec.http.HttpServerCodec;
  12. import io.netty.handler.logging.LogLevel;
  13. import io.netty.handler.logging.LoggingHandler;
  14.  
  15. public class TestHttp {
  16.     public static void main(String[] args) {
  17.         NioEventLoopGroup boss=new NioEventLoopGroup();
  18.         NioEventLoopGroup worker=new NioEventLoopGroup();
  19.         try{
  20.             ServerBootstrap serverBootstrap=new ServerBootstrap();
  21.             serverBootstrap.channel(NioServerSocketChannel.class);
  22.             serverBootstrap.group(boss,worker);
  23.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  24.                 @Override
  25.                 protected void initChannel(SocketChannel socketChannel) throws Exception {
  26.                     socketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  27.                     //Http的解码器
  28.                     socketChannel.pipeline().addLast(new HttpServerCodec());
  29.                     socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
  30.                         @Override
  31.                         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  32.                             //经过http解码器解码后，请求会被解析为请求头实体或者请求体实体
  33.                             if(msg instanceof HttpRequest){
  34.                                 //请求行、请求头
  35.                             }else if(msg instanceof HttpContent){
  36.                                 //请求体
  37.                             }
  38.                         }
  39.                     });
  40.                 }
  41.             });
  42.             ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
  43.             channelFuture.channel().closeFuture().sync();
  44.         }catch (Exception e){
  45.             e.printStackTrace();
  46.         }finally {
  47.             boss.shutdownGracefully();
  48.             worker.shutdownGracefully();
  49.         }
  50.     }
  51. }

当然上面这种写法太繁琐了，netty提供了SimpleChannelInboundHandler，用泛型来指定处理请求头还是请求体:

  1. import io.netty.bootstrap.ServerBootstrap;
  2. import io.netty.channel.ChannelFuture;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.http.*;
  10. import io.netty.handler.logging.LogLevel;
  11. import io.netty.handler.logging.LoggingHandler;
  12.  
  13. public class TestHttp {
  14.     public static void main(String[] args) {
  15.         NioEventLoopGroup boss=new NioEventLoopGroup();
  16.         NioEventLoopGroup worker=new NioEventLoopGroup();
  17.         try{
  18.             ServerBootstrap serverBootstrap=new ServerBootstrap();
  19.             serverBootstrap.channel(NioServerSocketChannel.class);
  20.             serverBootstrap.group(boss,worker);
  21.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  22.                 @Override
  23.                 protected void initChannel(SocketChannel socketChannel) throws Exception {
  24.                     socketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  25.                     //Http的解码器
  26.                     socketChannel.pipeline().addLast(new HttpServerCodec());
  27.                     socketChannel.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
  28.                         @Override
  29.                         protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  30.                             DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
  31.                             byte[] bytes = "<h1>hello world!</h1>".getBytes();
  32.                             //响应头设置返回的消息的长度，否则浏览器不知道消息有多长，会一直刷新
  33.                             response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH,bytes.length);
  34.                             //响应体设置返回的消息
  35.                             response.content().writeBytes(bytes);
  36.                             //写回响应
  37.                             ctx.writeAndFlush(response);
  38.                         }
  39.                     });
  40.                 }
  41.             });
  42.             ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
  43.             channelFuture.channel().closeFuture().sync();
  44.         }catch (Exception e){
  45.             e.printStackTrace();
  46.         }finally {
  47.             boss.shutdownGracefully();
  48.             worker.shutdownGracefully();
  49.         }
  50.     }
  51. }

12.协议设计

12.1.概述

自定义协议要素：

  • 魔术，用来判断数据包是否有效。

  • 版本号，协议版本号，用来支持协议升级。

  • 序列化算法，消息正文采用的序列化方式。

  • 指令类型，是登录、注册、还是其他........

  • 请求序号，用来支持双工通信，如TCP之类的。

  • 正文长度

  • 消息正文

12.2.编码

编解码，netty自带编解码器接口ByteToMessageCodec，允许开发者将数据报文转为自己想要的类型。

注意：想要转为的目标类型，必须是实现了序列化接口，可序列化的，不然会报错。

  1. public class MyCodec extends ByteToMessageCodec<Message> {
  2.     @Override
  3.     protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
  4.         //4字节的魔数
  5.         out.writeBytes(new byte[]{1, 2,3,4});
  6.         //1字节的版本
  7.         out.writeByte(1);
  8.         //1字节的序列化方式jdk
  9.         out.writeByte(0);
  10.         //1字节的指令类型
  11.         out.writeByte(msg.getMessageType());
  12.         //4个字节序号
  13.         out.writeInt(msg.getSequenceId());
  14.         //填充字段
  15.         out.writeByte(0xff);
  16.         //消息内容
  17.         ByteArrayOutputStream bos = new ByteArrayOutputStream();
  18.         ObjectOutputStream oos = new ObjectOutputStream(bos);
  19.         oos.writeObject(msg);byte[] bytes = bos.toByteArray();
  20.         //消息长度
  21.         out.writeInt(bytes.length);
  22.         //写入内容
  23.         out.writeBytes(bytes);
  24.     }
  25.  
  26.     @Override
  27.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  28.         //魔数
  29.         int magicNum = in.readInt();
  30.         //版本号
  31.         byte version = in.readByte();
  32.         //序列化类型
  33.         byte serializerType = in.readByte();
  34.         //指令类型
  35.         byte messageType = in.readByte();
  36.         //序号
  37.         int sequenceId = in.readInt();
  38.         //读填充字节
  39.         in.readByte();
  40.         //消息长度
  41.         int length = in.readInt();
  42.         //读消息
  43.         byte[] bytes = new byte[length];
  44.         in.readBytes(bytes, 0,length);
  45.         ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
  46.         Message message = (Message) ois.readObject();
  47.         out.add(message);
  48.     }
  49. }

测试：

  1. public static void main(String[] args) throws Exception {
  2.         EmbeddedChannel channel=new EmbeddedChannel(
  3.                 new LoggingHandler(),
  4.                 new MyCodec()
  5.         );
  6.         Message message=new Message();
  7.         message.setData("hello".getBytes());
  8.         //出站会调用codec的encode()
  9.         channel.writeOutbound(message);
  10.  
  11.         ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
  12.         new MyCodec().encode( null,message, buf);
  13.  
  14.         //入站会调用codec的decode()
  15.         channel.writeInbound(buf);
  16.     }

测试半包、粘包问题：

  1. public static void main(String[] args) throws Exception {
  2.         EmbeddedChannel channel=new EmbeddedChannel(
  3.                 new LengthFieldBasedFrameDecoder(1024,12,4,0,0),
  4.                 new LoggingHandler(),
  5.                 new MyCodec()
  6.         );
  7.         Message message=new Message();
  8.         message.setData("hello".getBytes());
  9.  
  10.         ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
  11.         new MyCodec().encode( null,message, buf);
  12.  
  13.         ByteBuf s1=buf.slice(0,100);
  14.         ByteBuf s2=buf.slice(100,buf.readableBytes()-100);
  15.  
  16.         //writeInbound后ByteBuf的引用计数会被-1，导致ByteBuf被释放掉，这里需要手动维持一下
  17.         s1.retain();
  18.         channel.writeInbound(s1);
  19.         channel.writeInbound(s2);
  20.     }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/812828
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号