Netty高级使用与源码详解

粘包与半包

粘包现象

粘包的问题出现是因为不知道一个用户消息的边界在哪，如果知道了边界在哪，接收方就可以通过边界来划分出有效的用户消息。

服务端代码

public class HelloWorldServer {
 static final Logger log = LoggerFactory.getLogger(HelloWorldServer.class);
 void start() {
 NioEventLoopGroup boss = new NioEventLoopGroup(1);
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 ServerBootstrap serverBootstrap = new ServerBootstrap();
 serverBootstrap.channel(NioServerSocketChannel.class);
 serverBootstrap.group(boss, worker);
 serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("connected {}", ctx.channel());
 super.channelActive(ctx);
 }
 @Override
 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 log.debug("disconnect {}", ctx.channel());
 super.channelInactive(ctx);
 }
 });
 }
 });
 ChannelFuture channelFuture = serverBootstrap.bind(8080);
 log.debug("{} binding...", channelFuture.channel());
 channelFuture.sync();
 log.debug("{} bound...", channelFuture.channel());
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("server error", e);
 } finally {
 boss.shutdownGracefully();
 worker.shutdownGracefully();
 log.debug("stoped");
 }
 }
 public static void main(String[] args) {
 new HelloWorldServer().start();
 }
}

客户端代码希望发送 10 个消息，每个消息是 16 字节

public class HelloWorldClient {
 static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
 public static void main(String[] args) {
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 log.debug("connetted...");
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 //会在连接channel建立成功后，会触发Active事件
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("sending...");
 Random r = new Random();
 char c = 'a';
 for (int i = 0; i < 10; i++) {
 ByteBuf buffer = ctx.alloc().buffer();
 buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
 ctx.writeAndFlush(buffer);
 }
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("client error", e);
 } finally {
 worker.shutdownGracefully();
 }
 }
}

服务器端的某次输出，可以看到一次就接收了 160 个字节，而期望的是一次16字节，分 10 次接收。这就出现了粘包现象

08:24:46 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x81e0fda5] binding...
08:24:46 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x81e0fda5, L:/0:0:0:0:0:0:0:0:8080] bound...
08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] REGISTERED
08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] ACTIVE
08:24:55 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177]
08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] READ: 160B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000020| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000030| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000040| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000050| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000060| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000070| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000080| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000090| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
+--------+-------------------------------------------------+----------------+
08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] READ COMPLETE

半包现象

半包是指接收端只收到了部分数据，而非完整的数据的情况

客户端代码希望发送 1 个消息，这个消息是 160 字节，代码改为

ByteBuf buffer = ctx.alloc().buffer();
for (int i = 0; i < 10; i++) {
 buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
}
ctx.writeAndFlush(buffer);

为现象明显，服务端修改一下接收缓冲区，其它代码不变

serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);

服务器端的某次输出，可以看到接收的消息被分为两节，如第一次 20 字节，第二次 140 字节

08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84] binding...
08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84, L:/0:0:0:0:0:0:0:0:8080] bound...
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] REGISTERED
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] ACTIVE
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221]
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 20B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03 |.... |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 140B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000010| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000020| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000030| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000040| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000050| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000060| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000070| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000080| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |............ |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE

注意：serverBootstrap.option(ChannelOption.SO_RCVBUF, 10) 影响的底层接收缓冲区（即滑动窗口）大小，仅决定了 netty 读取的最小单位，netty 实际每次读取的一般是它的整数倍

现象分析

这里出现的粘包半包问题，并非是JavaNIO或Netty的问题，本质是TCP是流失协议，消息无边界。

粘包：

现象，发送 abc def，接收 abcdef
原因
- 应用层：接收方 ByteBuf 设置太大（Netty 默认 1024）
- 滑动窗口：假设发送方 256 bytes 表示一个完整报文，但由于接收方处理不及时且窗口大小足够大，这 256 bytes 字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包
- Nagle 算法：会造成粘包

半包

现象，发送 abcdef，接收 abc def
原因
- 应用层：接收方 ByteBuf 小于实际发送数据量
- 滑动窗口：假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时放不下了，只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包
- MSS 限制：当发送的数据超过 MSS 限制后，会将数据切分发送，就会造成半包

解决方案

接下来看下Netty如何解决以上问题的：

短链接，发一个包建立一次连接，这样连接建立到连接断开之间就是消息的边界，缺点效率太低
每一条消息采用固定长度，缺点浪费空间
每一条消息采用分隔符，例如 \n，缺点需要转义
每一条消息分为 head 和 body，head 中包含 body 的长度

方法1：短链接（极不推荐）

以解决粘包为例

public class HelloWorldClient {
 static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
 public static void main(String[] args) {
 // 分 10 次发送
 for (int i = 0; i < 10; i++) {
 send();
 }
 }
 private static void send() {
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 log.debug("conneted...");
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("sending...");
 ByteBuf buffer = ctx.alloc().buffer();
 buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
 ctx.writeAndFlush(buffer);
 // 发完即关
 ctx.close();
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("client error", e);
 } finally {
 worker.shutdownGracefully();
 }
 }
}

输出，略

半包用这种办法还是不好解决，因为接收方的缓冲区大小是有限的

方法2：固定长度

让所有数据包长度固定（假设长度为 8 字节），服务器端加入

ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

客户端测试代码，注意, 采用这种方法后，客户端什么时候 flush 都可以

public class HelloWorldClient {
 static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
 public static void main(String[] args) {
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 log.debug("connetted...");
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("sending...");
 // 发送内容随机的数据包
 Random r = new Random();
 char c = 'a';
 ByteBuf buffer = ctx.alloc().buffer();
 for (int i = 0; i < 10; i++) {
 byte[] bytes = new byte[8];
 for (int j = 0; j < r.nextInt(8); j++) {
 bytes[j] = (byte) c;
 }
 c++;
 buffer.writeBytes(bytes);
 }
 ctx.writeAndFlush(buffer);
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("client error", e);
 } finally {
 worker.shutdownGracefully();
 }
 }
}

客户端输出

12:07:00 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2] REGISTERED
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2] CONNECT: /192.168.0.103:9090
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/192.168.0.103:53155 - R:/192.168.0.103:9090] ACTIVE
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/192.168.0.103:53155 - R:/192.168.0.103:9090] WRITE: 80B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 00 00 00 00 62 00 00 00 00 00 00 00 |aaaa....b.......|
|00000010| 63 63 00 00 00 00 00 00 64 00 00 00 00 00 00 00 |cc......d.......|
|00000020| 00 00 00 00 00 00 00 00 66 66 66 66 00 00 00 00 |........ffff....|
|00000030| 67 67 67 00 00 00 00 00 68 00 00 00 00 00 00 00 |ggg.....h.......|
|00000040| 69 69 69 69 69 00 00 00 6a 6a 6a 6a 00 00 00 00 |iiiii...jjjj....|
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/192.168.0.103:53155 - R:/192.168.0.103:9090] FLUSH

服务端输出

12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f] binding...
12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f, L:/192.168.0.103:9090] bound...
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] REGISTERED
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] ACTIVE
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155]
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 00 00 00 00 |aaaa.... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 00 00 00 00 00 00 00 |b....... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 63 00 00 00 00 00 00 |cc...... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 64 00 00 00 00 00 00 00 |d....... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 00 00 00 00 00 |........ |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 66 66 66 00 00 00 00 |ffff.... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 67 67 67 00 00 00 00 00 |ggg..... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 00 00 00 00 00 00 00 |h....... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 69 69 69 69 69 00 00 00 |iiiii... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6a 6a 6a 6a 00 00 00 00 |jjjj.... |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ COMPLETE

缺点是，数据包的大小不好把握

长度定的太大，浪费
长度定的太小，对某些数据包又显得不够

方法3：固定分隔符

服务端加入，默认以 \n 或 \r\n 作为分隔符，如果超出指定长度仍未出现分隔符，则抛出异常

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

客户端在每条消息之后，加入 \n 分隔符

public class HelloWorldClient {
 static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
 public static void main(String[] args) {
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 log.debug("connetted...");
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("sending...");
 Random r = new Random();
 char c = 'a';
 ByteBuf buffer = ctx.alloc().buffer();
 for (int i = 0; i < 10; i++) {
 for (int j = 1; j <= r.nextInt(16)+1; j++) {
 buffer.writeByte((byte) c);
 }
 buffer.writeByte(10);
 c++;
 }
 ctx.writeAndFlush(buffer);
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("client error", e);
 } finally {
 worker.shutdownGracefully();
 }
 }
}

客户端输出

14:08:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755] REGISTERED
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755] CONNECT: /192.168.0.103:9090
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] ACTIVE
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] WRITE: 60B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 0a 62 62 62 0a 63 63 63 0a 64 64 0a 65 65 65 |a.bbb.ccc.dd.eee|
|00000010| 65 65 65 65 65 65 65 0a 66 66 0a 67 67 67 67 67 |eeeeeee.ff.ggggg|
|00000020| 67 67 0a 68 68 68 68 0a 69 69 69 69 69 69 69 0a |gg.hhhh.iiiiiii.|
|00000030| 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 0a |jjjjjjjjjjj. |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] FLUSH

服务端输出

14:08:18 [DEBUG] [nioEventLoopGroup-3-5] c.i.n.HelloWorldServer - connected [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641]
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 1B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 |a |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 3B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 62 |bbb |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 3B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 63 63 |ccc |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 2B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 64 64 |dd |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 10B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeee |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 2B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 66 |ff |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 7B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 67 67 67 67 67 67 67 |ggggggg |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 4B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 68 68 68 |hhhh |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 7B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 69 69 69 69 69 69 69 |iiiiiii |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 11B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjjjj |
+--------+-------------------------------------------------+----------------+
14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ COMPLETE

缺点，处理字符数据比较合适，但如果内容本身包含了分隔符（字节数据常常会有此情况），那么就会解析错误

方法4：预设长度

在发送消息前，先约定用定长字节表示接下来数据的长度

// 最大长度，长度偏移，长度占用字节，长度调整，剥离字节数
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1));

客户端代码

public class HelloWorldClient {
 static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
 public static void main(String[] args) {
 NioEventLoopGroup worker = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 log.debug("connetted...");
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 log.debug("sending...");
 Random r = new Random();
 char c = 'a';
 ByteBuf buffer = ctx.alloc().buffer();
 for (int i = 0; i < 10; i++) {
 byte length = (byte) (r.nextInt(16) + 1);
 // 先写入长度
 buffer.writeByte(length);
 // 再
 for (int j = 1; j <= length; j++) {
 buffer.writeByte((byte) c);
 }
 c++;
 }
 ctx.writeAndFlush(buffer);
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
 channelFuture.channel().closeFuture().sync();
 } catch (InterruptedException e) {
 log.error("client error", e);
 } finally {
 worker.shutdownGracefully();
 }
 }
}

客户端输出

14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] REGISTERED
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] CONNECT: /192.168.0.103:9090
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] ACTIVE
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] WRITE: 97B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 09 61 61 61 61 61 61 61 61 61 09 62 62 62 62 62 |.aaaaaaaaa.bbbbb|
|00000010| 62 62 62 62 06 63 63 63 63 63 63 08 64 64 64 64 |bbbb.cccccc.dddd|
|00000020| 64 64 64 64 0f 65 65 65 65 65 65 65 65 65 65 65 |dddd.eeeeeeeeeee|
|00000030| 65 65 65 65 0d 66 66 66 66 66 66 66 66 66 66 66 |eeee.fffffffffff|
|00000040| 66 66 02 67 67 02 68 68 0e 69 69 69 69 69 69 69 |ff.gg.hh.iiiiiii|
|00000050| 69 69 69 69 69 69 69 09 6a 6a 6a 6a 6a 6a 6a 6a |iiiiiii.jjjjjjjj|
|00000060| 6a |j |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] FLUSH

服务端输出

14:36:50 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3] binding...
14:36:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3, L:/192.168.0.103:9090] bound...
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] REGISTERED
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] ACTIVE
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979]
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 |aaaaaaaaa |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 62 62 62 62 62 62 62 |bbbbbbbbb |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 6B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 63 63 63 63 63 |cccccc |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 8B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 64 64 64 64 64 64 64 64 |dddddddd |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 15B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 65 65 65 65 65 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeeeeeeee |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 13B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 66 66 66 66 66 66 66 66 66 66 66 66 |fffffffffffff |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 2B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 67 67 |gg |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 2B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 68 |hh |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 14B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 69 69 69 69 69 69 69 69 69 69 69 69 69 69 |iiiiiiiiiiiiii |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
 +-------------------------------------------------+
 | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjj |
+--------+-------------------------------------------------+----------------+
14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ COMPLETE

协议设计与解析

为什么需要协议？

TCP/IP 中消息传输基于流的方式，没有边界。

协议的目的就是划定消息的边界，制定通信双方要共同遵守的通信规则

例如：在网络上传输

下雨天留客天留我不留

是中文一句著名的无标点符号句子，在没有标点符号情况下，这句话有数种拆解方式，而意思却是完全不同，所以常被用作讲述标点符号的重要性

一种解读

下雨天留客，天留，我不留

另一种解读

下雨天，留客天，留我不？留

如何设计协议呢？其实就是给网络传输的信息加上“标点符号”。但通过分隔符来断句不是很好，因为分隔符本身如果用于传输，那么必须加以区分。因此，下面一种协议较为常用

定长字节表示内容长度 + 实际内容

例如，假设一个中文字符长度为 3，按照上述协议的规则，发送信息方式如下，就不会被接收方弄错意思了

0f下雨天留客06天留09我不留

redis 协议举例

模拟 redis 客户端发送命令。

NioEventLoopGroup worker = new NioEventLoopGroup();
byte[] LINE = {13, 10};
try {
 Bootstrap bootstrap = new Bootstrap();
 bootstrap.channel(NioSocketChannel.class);
 bootstrap.group(worker);
 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) {
 ch.pipeline().addLast(new LoggingHandler());
 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 // 会在连接 channel 建立成功后，会触发 active 事件
 @Override
 public void channelActive(ChannelHandlerContext ctx) {
 set(ctx);
 get(ctx);
 }
 private void get(ChannelHandlerContext ctx) {
 ByteBuf buf = ctx.alloc().buffer();
 buf.writeBytes("*2".getBytes());//*2 表示数组元素个数为2，即get aaa 是2串内容
 buf.writeBytes(LINE);
 buf.writeBytes("$3".getBytes());//规定用 $3 表示后续有3个字节
 buf.writeBytes(LINE);
 buf.writeBytes("get".getBytes());//输入gset命令
 buf.writeBytes(LINE);
 buf.writeBytes("$3".getBytes());
 buf.writeBytes(LINE);
 buf.writeBytes("aaa".getBytes());//输入key为 aaa 
 buf.writeBytes(LINE);
 ctx.writeAndFlush(buf);
 }
 private void set(ChannelHandlerContext ctx) {
 //以下redis命令为 set aaa bbb
 ByteBuf buf = ctx.alloc().buffer();
 buf.writeBytes("*3".getBytes());//*3 表示数组元素个数为3，即set aaa bbb是3串内容
 buf.writeBytes(LINE);
 buf.writeBytes("$3".getBytes());//规定用 $3 表示后续有3个字节
 buf.writeBytes(LINE);
 buf.writeBytes("set".getBytes());//输入set命令
 buf.writeBytes(LINE);
 buf.writeBytes("$3".getBytes());
 buf.writeBytes(LINE);
 buf.writeBytes("aaa".getBytes());//输入key为 aaa 
 buf.writeBytes(LINE);
 buf.writeBytes("$3".getBytes());
 buf.writeBytes(LINE);
 buf.writeBytes("bbb".getBytes());//输入value为 bbb
 buf.writeBytes(LINE);
 ctx.writeAndFlush(buf);
 }
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 ByteBuf buf = (ByteBuf) msg;
 System.out.println(buf.toString(Charset.defaultCharset()));
 }
 });
 }
 });
 ChannelFuture channelFuture = bootstrap.connect("localhost", 6379).sync();
 channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
 log.error("client error", e);
} finally {
 worker.shutdownGracefully();
}

当然 netty提供了现成的这些协议，不需要我们自己来开发，这里是为了知其所以然

http 协议举例

模拟http服务端

NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
 ServerBootstrap serverBootstrap = new ServerBootstrap();
 serverBootstrap.channel(NioServerSocketChannel.class);
 serverBootstrap.group(boss, worker);
 serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
 ch.pipeline().addLast(new HttpServerCodec());
 ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
 // 获取请求
 log.debug(msg.uri());
 // 返回响应
 DefaultFullHttpResponse response =
 new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
 byte[] bytes = "<h1>Hello, world!</h1>".getBytes();
 response.headers().setInt(CONTENT_LENGTH, bytes.length);
 response.content().writeBytes(bytes);
 // 写回响应
 ctx.writeAndFlush(response);
 }
 });
 /*ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 log.debug("{}", msg.getClass());
 if (msg instanceof HttpRequest) { // 请求行，请求头
 } else if (msg instanceof HttpContent) { //请求体
 }
 }
 });*/
 }
 });
 ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
 channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
 log.error("server error", e);
} finally {
 boss.shutdownGracefully();
 worker.shutdownGracefully();
}

自定义协议要素

魔数：约定好的，用来在第一时间判定是否是无效数据包。
版本号：可以支持协议的升级
序列化算法：消息正文到底采用哪种序列化反序列化方式，可以由此扩展，例如：json、protobuf、hessian、jdk
指令类型：是登录、注册、单聊、群聊... 跟业务相关
请求序号：为了双工通信，提供异步能力
正文长度
消息正文

编解码器

根据上面的要素，设计一个登录请求消息和登录响应消息，并使用 Netty 完成收发

@Slf4j
public class MessageCodec extends ByteToMessageCodec<Message> {
 @Override
 protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
 // 1. 4 字节的魔数
 out.writeBytes(new byte[]{1, 2, 3, 4});
 // 2. 1 字节的版本,
 out.writeByte(1);
 // 3. 1 字节的序列化方式 jdk 0 , json 1
 out.writeByte(0);
 // 4. 1 字节的指令类型
 out.writeByte(msg.getMessageType());
 // 5. 4 个字节
 out.writeInt(msg.getSequenceId());
 // 无意义，对齐填充
 out.writeByte(0xff);
 // 6. 获取内容的字节数组
 ByteArrayOutputStream bos = new ByteArrayOutputStream();
 ObjectOutputStream oos = new ObjectOutputStream(bos);
 oos.writeObject(msg);
 byte[] bytes = bos.toByteArray();
 // 7. 长度
 out.writeInt(bytes.length);
 // 8. 写入内容
 out.writeBytes(bytes);
 }
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
 int magicNum = in.readInt();
 byte version = in.readByte();
 byte serializerType = in.readByte();
 byte messageType = in.readByte();
 int sequenceId = in.readInt();
 in.readByte();
 int length = in.readInt();
 byte[] bytes = new byte[length];
 in.readBytes(bytes, 0, length);
 ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
 Message message = (Message) ois.readObject();
 log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
 log.debug("{}", message);
 out.add(message);
 }
}

测试

EmbeddedChannel channel = new EmbeddedChannel(
 new LoggingHandler(),
 new LengthFieldBasedFrameDecoder(
 1024, 12, 4, 0, 0),
 new MessageCodec()
);
// encode
LoginRequestMessage message = new LoginRequestMessage("zhangsan", "123", "张三");
// channel.writeOutbound(message);
// decode
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
new MessageCodec().encode(null, message, buf);
ByteBuf s1 = buf.slice(0, 100);
ByteBuf s2 = buf.slice(100, buf.readableBytes() - 100);
s1.retain(); // 引用计数 2
channel.writeInbound(s1); // release 1
channel.writeInbound(s2);

@Sharable

当 handler 不保存状态时，就可以安全地在多线程下被共享
但要注意对于编解码器类，不能继承 ByteToMessageCodec 或 CombinedChannelDuplexHandler 父类，他们的构造方法对 @Sharable 有限制
如果能确保编解码器不会保存状态，可以继承 MessageToMessageCodec 父类

@Slf4j
@ChannelHandler.Sharable
/**
 * 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的
 */
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
 @Override
 protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
 ByteBuf out = ctx.alloc().buffer();
 // 1. 4 字节的魔数
 out.writeBytes(new byte[]{1, 2, 3, 4});
 // 2. 1 字节的版本,
 out.writeByte(1);
 // 3. 1 字节的序列化方式 jdk 0 , json 1
 out.writeByte(0);
 // 4. 1 字节的指令类型
 out.writeByte(msg.getMessageType());
 // 5. 4 个字节
 out.writeInt(msg.getSequenceId());
 // 无意义，对齐填充
 out.writeByte(0xff);
 // 6. 获取内容的字节数组
 ByteArrayOutputStream bos = new ByteArrayOutputStream();
 ObjectOutputStream oos = new ObjectOutputStream(bos);
 oos.writeObject(msg);
 byte[] bytes = bos.toByteArray();
 // 7. 长度
 out.writeInt(bytes.length);
 // 8. 写入内容
 out.writeBytes(bytes);
 outList.add(out);
 }
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
 int magicNum = in.readInt();
 byte version = in.readByte();
 byte serializerType = in.readByte();
 byte messageType = in.readByte();
 int sequenceId = in.readInt();
 in.readByte();
 int length = in.readInt();
 byte[] bytes = new byte[length];
 in.readBytes(bytes, 0, length);
 ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
 Message message = (Message) ois.readObject();
 log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
 log.debug("{}", message);
 out.add(message);
 }
}

扩展序列化算法

序列化，反序列化主要用在消息正文的转换上

序列化时，需要将 Java 对象变为要传输的数据（可以是 byte[]，或 json 等，最终都需要变成 byte[]）
反序列化时，需要将传入的正文数据还原成 Java 对象，便于处理

目前的代码仅支持 Java 自带的序列化，反序列化机制，核心代码如下

// 反序列化
byte[] body = new byte[bodyLength];
byteByf.readBytes(body);
ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(body));
Message message = (Message) in.readObject();
message.setSequenceId(sequenceId);
// 序列化
ByteArrayOutputStream out = new ByteArrayOutputStream();
new ObjectOutputStream(out).writeObject(message);
byte[] bytes = out.toByteArray();

为了支持更多序列化算法，抽象一个 Serializer 接口

public interface Serializer {
 // 反序列化方法
 <T> T deserialize(Class<T> clazz, byte[] bytes);
 // 序列化方法
 <T> byte[] serialize(T object);
}

提供两个实现，这里直接将具体实现加入了枚举类 Serializer.Algorithm 中

enum SerializerAlgorithm implements Serializer {
	// Java 实现
 Java {
 @Override
 public <T> T deserialize(Class<T> clazz, byte[] bytes) {
 try {
 ObjectInputStream in = 
 new ObjectInputStream(new ByteArrayInputStream(bytes));
 Object object = in.readObject();
 return (T) object;
 } catch (IOException | ClassNotFoundException e) {
 throw new RuntimeException("SerializerAlgorithm.Java 反序列化错误", e);
 }
 }
 @Override
 public <T> byte[] serialize(T object) {
 try {
 ByteArrayOutputStream out = new ByteArrayOutputStream();
 new ObjectOutputStream(out).writeObject(object);
 return out.toByteArray();
 } catch (IOException e) {
 throw new RuntimeException("SerializerAlgorithm.Java 序列化错误", e);
 }
 }
 }, 
 
 // Json 实现(引入了 Gson 依赖)
 Json {
 @Override
 public <T> T deserialize(Class<T> clazz, byte[] bytes) {
 return new Gson().fromJson(new String(bytes, StandardCharsets.UTF_8), clazz);
 }
 @Override
 public <T> byte[] serialize(T object) {
 return new Gson().toJson(object).getBytes(StandardCharsets.UTF_8);
 }
 };
 // 需要从协议的字节中得到是哪种序列化算法
 public static SerializerAlgorithm getByInt(int type) {
 SerializerAlgorithm[] array = SerializerAlgorithm.values();
 if (type < 0 || type > array.length - 1) {
 throw new IllegalArgumentException("超过 SerializerAlgorithm 范围");
 }
 return array[type];
 }
}

增加配置类和配置文件

public abstract class Config {
 static Properties properties;
 static {
 try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
 properties = new Properties();
 properties.load(in);
 } catch (IOException e) {
 throw new ExceptionInInitializerError(e);
 }
 }
 public static int getServerPort() {
 String value = properties.getProperty("server.port");
 if(value == null) {
 return 8080;
 } else {
 return Integer.parseInt(value);
 }
 }
 public static Serializer.Algorithm getSerializerAlgorithm() {
 String value = properties.getProperty("serializer.algorithm");
 if(value == null) {
 return Serializer.Algorithm.Java;
 } else {
 return Serializer.Algorithm.valueOf(value);
 }
 }
}

配置文件

serializer.algorithm=Json

修改编解码器

/**
 * 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的
 */
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
 @Override
 public void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
 ByteBuf out = ctx.alloc().buffer();
 // 1. 4 字节的魔数
 out.writeBytes(new byte[]{1, 2, 3, 4});
 // 2. 1 字节的版本,
 out.writeByte(1);
 // 3. 1 字节的序列化方式 jdk 0 , json 1
 out.writeByte(Config.getSerializerAlgorithm().ordinal());
 // 4. 1 字节的指令类型
 out.writeByte(msg.getMessageType());
 // 5. 4 个字节
 out.writeInt(msg.getSequenceId());
 // 无意义，对齐填充
 out.writeByte(0xff);
 // 6. 获取内容的字节数组
 byte[] bytes = Config.getSerializerAlgorithm().serialize(msg);
 // 7. 长度
 out.writeInt(bytes.length);
 // 8. 写入内容
 out.writeBytes(bytes);
 outList.add(out);
 }
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
 int magicNum = in.readInt();
 byte version = in.readByte();
 byte serializerAlgorithm = in.readByte(); // 0 或 1
 byte messageType = in.readByte(); // 0,1,2...
 int sequenceId = in.readInt();
 in.readByte();
 int length = in.readInt();
 byte[] bytes = new byte[length];
 in.readBytes(bytes, 0, length);
 // 找到反序列化算法
 Serializer.Algorithm algorithm = Serializer.Algorithm.values()[serializerAlgorithm];
 // 确定具体消息类型
 Class<? extends Message> messageClass = Message.getMessageClass(messageType);
 Message message = algorithm.deserialize(messageClass, bytes);
// log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
// log.debug("{}", message);
 out.add(message);
 }
}

其中确定具体消息类型，可以根据 消息类型字节 获取到对应的 消息 class

@Data
public abstract class Message implements Serializable {
 /**
 * 根据消息类型字节，获得对应的消息 class
 * @param messageType 消息类型字节
 * @return 消息 class
 */
 public static Class<? extends Message> getMessageClass(int messageType) {
 return messageClasses.get(messageType);
 }
 private int sequenceId;
 private int messageType;
 public abstract int getMessageType();
 public static final int LoginRequestMessage = 0;
 public static final int LoginResponseMessage = 1;
 public static final int ChatRequestMessage = 2;
 public static final int ChatResponseMessage = 3;
 public static final int GroupCreateRequestMessage = 4;
 public static final int GroupCreateResponseMessage = 5;
 public static final int GroupJoinRequestMessage = 6;
 public static final int GroupJoinResponseMessage = 7;
 public static final int GroupQuitRequestMessage = 8;
 public static final int GroupQuitResponseMessage = 9;
 public static final int GroupChatRequestMessage = 10;
 public static final int GroupChatResponseMessage = 11;
 public static final int GroupMembersRequestMessage = 12;
 public static final int GroupMembersResponseMessage = 13;
 public static final int PingMessage = 14;
 public static final int PongMessage = 15;
 private static final Map<Integer, Class<? extends Message>> messageClasses = new HashMap<>();
 static {
 messageClasses.put(LoginRequestMessage, LoginRequestMessage.class);
 messageClasses.put(LoginResponseMessage, LoginResponseMessage.class);
 messageClasses.put(ChatRequestMessage, ChatRequestMessage.class);
 messageClasses.put(ChatResponseMessage, ChatResponseMessage.class);
 messageClasses.put(GroupCreateRequestMessage, GroupCreateRequestMessage.class);
 messageClasses.put(GroupCreateResponseMessage, GroupCreateResponseMessage.class);
 messageClasses.put(GroupJoinRequestMessage, GroupJoinRequestMessage.class);
 messageClasses.put(GroupJoinResponseMessage, GroupJoinResponseMessage.class);
 messageClasses.put(GroupQuitRequestMessage, GroupQuitRequestMessage.class);
 messageClasses.put(GroupQuitResponseMessage, GroupQuitResponseMessage.class);
 messageClasses.put(GroupChatRequestMessage, GroupChatRequestMessage.class);
 messageClasses.put(GroupChatResponseMessage, GroupChatResponseMessage.class);
 messageClasses.put(GroupMembersRequestMessage, GroupMembersRequestMessage.class);
 messageClasses.put(GroupMembersResponseMessage, GroupMembersResponseMessage.class);
 }
}

参数调优

相关源码待更新

客户端参数 CONNECT_TIMEOUT_MILLIS

属于 SocketChannal 参数
用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常
SO_TIMEOUT 主要用在阻塞 IO，阻塞 IO 中 accept，read 等都是无限等待的，如果不希望永远阻塞，使用它调整超时时间

@Slf4j
public class TestConnectionTimeout {
 public static void main(String[] args) {
 NioEventLoopGroup group = new NioEventLoopGroup();
 try {
 Bootstrap bootstrap = new Bootstrap()
 .group(group)
 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 300)
 .channel(NioSocketChannel.class)
 .handler(new LoggingHandler());
 ChannelFuture future = bootstrap.connect("127.0.0.1", 8080);
 future.sync().channel().closeFuture().sync(); // 断点1
 } catch (Exception e) {
 e.printStackTrace();
 log.debug("timeout");
 } finally {
 group.shutdownGracefully();
 }
 }
}

另外源码部分 io.netty.channel.nio.AbstractNioChannel.AbstractNioUnsafe#connect

@Override
public final void connect(
 final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
 // ...
 // 获取超时时间
 int connectTimeoutMillis = config().getConnectTimeoutMillis();
 if (connectTimeoutMillis > 0) {//如果超时时间大于0
 connectTimeoutFuture = eventLoop().schedule(new Runnable() {//则启动定时任务
 @Override
 public void run() { 
 //connectPromise是两个线程间交换数据的对象
 ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
 ConnectTimeoutException cause =
 new ConnectTimeoutException("connection timed out: " + remoteAddress); // 断点2
 //connectPromise.tryFailure(cause) 将异常放到connectPromise里，再唤醒主线程
 if (connectPromise != null && connectPromise.tryFailure(cause)) {
 close(voidPromise());
 }
 }
 }, connectTimeoutMillis, TimeUnit.MILLISECONDS);//定时任务在 connectTimeoutMillis 时间后执行
 }
	// ...
}

服务端参数 SO_BACKLOG

这是属于 ServerSocketChannal 的参数

三次握手时有半连接队列和全连接队列，详情看这篇文章：TCP - 半连接队列和全连接队列

sync queue - 半连接队列
- 大小通过 /proc/sys/net/ipv4/tcp_max_syn_backlog 指定，在 syncookies 启用的情况下，逻辑上没有最大值限制，这个设置便被忽略
accept queue - 全连接队列
- 其大小通过 /proc/sys/net/core/somaxconn 指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值
- 如果 accpet queue 队列满了，server 将发送一个拒绝连接的错误信息到 client

netty 中可以通过 option(ChannelOption.SO_BACKLOG, 值) 来设置backlog 的大小

可以通过下面源码查看默认大小

public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
 implements ServerSocketChannelConfig {
 private volatile int backlog = NetUtil.SOMAXCONN;
 // ...
}

TCP_NODELAY

属于 SocketChannal 参数

立即发送，建议设置成true。false就是开启了nagle算法

SO_SNDBUF & SO_RCVBUF

设置滑动窗口的参数，在早些可能需要设置这些参数，但现在tcp会根据拥塞等对窗口进行自动调整，因此不建议手动设置这两个值。

SO_SNDBUF 属于 SocketChannal 参数
SO_RCVBUF 既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）

ALLOCATOR

属于 SocketChannal 参数
用来分配 ByteBuf， ctx.alloc()

RCVBUF_ALLOCATOR

属于 SocketChannal 参数
控制 netty 接收缓冲区大小
负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator 决定

源码详解

启动剖析

我们就来看看 netty 中对下面的代码是怎样进行处理的

//1 netty 中使用 NioEventLoopGroup （简称 nio boss 线程）来封装线程和 selector
Selector selector = Selector.open(); 
//2 创建 NioServerSocketChannel，同时会初始化它关联的 handler，以及为原生 ssc 存储 config
NioServerSocketChannel attachment = new NioServerSocketChannel();
//3 创建 NioServerSocketChannel 时，创建了 java 原生的 ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 
serverSocketChannel.configureBlocking(false);
//4 启动 nio boss 线程执行接下来的操作
//5 注册（仅关联 selector 和 NioServerSocketChannel），未关注事件
SelectionKey selectionKey = serverSocketChannel.register(selector, 0, attachment);
//6 head -> 初始化器 -> ServerBootstrapAcceptor -> tail，初始化器是一次性的，只为添加 acceptor
//7 绑定端口
serverSocketChannel.bind(new InetSocketAddress(8080));
//8 触发 channel active 事件，在 head 中关注 op_accept 事件
selectionKey.interestOps(SelectionKey.OP_ACCEPT);

入口 io.netty.bootstrap.ServerBootstrap#bind

关键代码 io.netty.bootstrap.AbstractBootstrap#doBind

这个函数是由哪些线程处理的呢？可以先有个概念，再往下看：

init & register regFuture 处理
1. init：由main处理
  1. 创建NioServerSocketChannel：由main处理
  2. 添加 NioServerSocketChannel 初始化 handler ：由main处理
    1. 初始化 handler 等待调用
register
1. 启动 nio boss 线程：由main处理
2. 原生 ssc 注册至 selector 未关注事件：由nio-thread处理
3. 执行 NioServerSocketChannel 初始化 handler：由nio-thread处理
regFuture 等待回调 doBind0：由nio-thread处理
1. 原生 ServerSocketChannel 绑定：由nio-thread处理
2. 触发NioServerSocketChannel active 事件：由nio-thread处理

private ChannelFuture doBind(final SocketAddress localAddress) {
	// 1. 执行初始化和注册 regFuture 会由 initAndRegister 设置其是否完成，从而回调 3.2 处代码
 // init就相当于 ServerSocketChannel ssc= ServerSocketChannel.open();
 // Register就相当于 SelectionKey selectionKey=ssc.register(selector, 0, nettySsc);
 final ChannelFuture regFuture = initAndRegister();
 final Channel channel = regFuture.channel();
 if (regFuture.cause() != null) {
 return regFuture;
 }
 // 2. 因为是 initAndRegister 异步执行，需要分两种情况来看，调试时也需要通过 suspend 断点类型加以区分
 // 2.1 如果已经完成
 if (regFuture.isDone()) {
 ChannelPromise promise = channel.newPromise();
 // 3.1 立刻调用 doBind0
 doBind0(regFuture, channel, localAddress, promise);
 return promise;
 } 
 // 2.2 还没有完成
 else {
 final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
 // 3.2 回调 doBind0
 regFuture.addListener(new ChannelFutureListener() {
 @Override
 public void operationComplete(ChannelFuture future) throws Exception {
 Throwable cause = future.cause();
 if (cause != null) {
 // 处理异常...
 promise.setFailure(cause);
 } else {
 promise.registered();
	// 3. 由注册线程去执行 doBind0
 doBind0(regFuture, channel, localAddress, promise);
 }
 }
 });
 return promise;
 }
}

关键代码 io.netty.bootstrap.AbstractBootstrap#initAndRegister

final ChannelFuture initAndRegister() {
 Channel channel = null;
 try {
 channel = channelFactory.newChannel();
 // 1.1 初始化 - 做的事就是添加一个初始化器 ChannelInitializer
 init(channel);
 } catch (Throwable t) {
 // 处理异常...
 return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
 }
 // 1.2 注册 - 做的事就是将原生 channel 注册到 selector 上
 ChannelFuture regFuture = config().group().register(channel);
 if (regFuture.cause() != null) {
 // 处理异常...
 }
 return regFuture;
}

关键代码 io.netty.bootstrap.ServerBootstrap#init

// 这里 channel 实际上是 NioServerSocketChannel
void init(Channel channel) throws Exception {
 final Map<ChannelOption<?>, Object> options = options0();
 synchronized (options) {
 setChannelOptions(channel, options, logger);
 }
 final Map<AttributeKey<?>, Object> attrs = attrs0();
 synchronized (attrs) {
 for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
 @SuppressWarnings("unchecked")
 AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
 channel.attr(key).set(e.getValue());
 }
 }
 ChannelPipeline p = channel.pipeline();
 final EventLoopGroup currentChildGroup = childGroup;
 final ChannelHandler currentChildHandler = childHandler;
 final Entry<ChannelOption<?>, Object>[] currentChildOptions;
 final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
 synchronized (childOptions) {
 currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
 }
 synchronized (childAttrs) {
 currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
 }
	
 // 为 NioServerSocketChannel 添加初始化器
 p.addLast(new ChannelInitializer<Channel>() {
 @Override
 public void initChannel(final Channel ch) throws Exception {
 final ChannelPipeline pipeline = ch.pipeline();
 ChannelHandler handler = config.handler();
 if (handler != null) {
 pipeline.addLast(handler);
 }
 // 初始化器的职责是将 ServerBootstrapAcceptor 加入至 NioServerSocketChannel
 ch.eventLoop().execute(new Runnable() {
 @Override
 public void run() {
 pipeline.addLast(new ServerBootstrapAcceptor(
 ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
 }
 });
 }
 });
}

关键代码 io.netty.channel.AbstractChannel.AbstractUnsafe#register

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
 // 一些检查，略...
 AbstractChannel.this.eventLoop = eventLoop;
 if (eventLoop.inEventLoop()) {
 register0(promise);
 } else {
 try {
 // 首次执行 execute 方法时，会启动 nio 线程，之后注册等操作在 nio 线程上执行
 // 因为只有一个 NioServerSocketChannel 因此，也只会有一个 boss nio 线程
 // 这行代码完成的事实是 main -> nio boss 线程的切换
 eventLoop.execute(new Runnable() {
 @Override
 public void run() {
 register0(promise);
 }
 });
 } catch (Throwable t) {
 // 日志记录...
 closeForcibly();
 closeFuture.setClosed();
 safeSetFailure(promise, t);
 }
 }
}

io.netty.channel.AbstractChannel.AbstractUnsafe#register0

private void register0(ChannelPromise promise) {
 try {
 if (!promise.setUncancellable() || !ensureOpen(promise)) {
 return;
 }
 boolean firstRegistration = neverRegistered;
 // 1.2.1 原生的 nio channel 绑定到 selector 上，注意此时没有注册 selector 关注事件，附件为 NioServerSocketChannel
 doRegister();
 neverRegistered = false;
 registered = true;
 // 1.2.2 执行 NioServerSocketChannel 初始化器的 initChannel
 pipeline.invokeHandlerAddedIfNeeded();
 // 回调 3.2 io.netty.bootstrap.AbstractBootstrap#doBind0
 safeSetSuccess(promise);
 pipeline.fireChannelRegistered();
 
 // 对应 server socket channel 还未绑定，isActive 为 false
 if (isActive()) {
 if (firstRegistration) {
 pipeline.fireChannelActive();
 } else if (config().isAutoRead()) {
 beginRead();
 }
 }
 } catch (Throwable t) {
 // Close the channel directly to avoid FD leak.
 closeForcibly();
 closeFuture.setClosed();
 safeSetFailure(promise, t);
 }
}

关键代码 io.netty.channel.ChannelInitializer#initChannel

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
 if (initMap.add(ctx)) { // Guard against re-entrance.
 try {
 // 1.2.2.1 执行初始化
 initChannel((C) ctx.channel());
 } catch (Throwable cause) {
 exceptionCaught(ctx, cause);
 } finally {
 // 1.2.2.2 移除初始化器
 ChannelPipeline pipeline = ctx.pipeline();
 if (pipeline.context(this) != null) {
 pipeline.remove(this);
 }
 }
 return true;
 }
 return false;
}

关键代码 io.netty.bootstrap.AbstractBootstrap#doBind0

// 3.1 或 3.2 执行 doBind0
private static void doBind0(
 final ChannelFuture regFuture, final Channel channel,
 final SocketAddress localAddress, final ChannelPromise promise) {
 channel.eventLoop().execute(new Runnable() {
 @Override
 public void run() {
 if (regFuture.isSuccess()) {
 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
 } else {
 promise.setFailure(regFuture.cause());
 }
 }
 });
}

关键代码 io.netty.channel.AbstractChannel.AbstractUnsafe#bind

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
 assertEventLoop();
 if (!promise.setUncancellable() || !ensureOpen(promise)) {
 return;
 }
 if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
 localAddress instanceof InetSocketAddress &&
 !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
 !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
 // 记录日志...
 }
 boolean wasActive = isActive();
 try {
 // 3.3 执行端口绑定
 doBind(localAddress);
 } catch (Throwable t) {
 safeSetFailure(promise, t);
 closeIfClosed();
 return;
 }
 if (!wasActive && isActive()) {
 invokeLater(new Runnable() {
 @Override
 public void run() {
 // 3.4 触发 active 事件
 pipeline.fireChannelActive();
 }
 });
 }
 safeSetSuccess(promise);
}

关键代码 io.netty.channel.socket.nio.NioServerSocketChannel#doBind

protected void doBind(SocketAddress localAddress) throws Exception {
 if (PlatformDependent.javaVersion() >= 7) {
 javaChannel().bind(localAddress, config.getBacklog());
 } else {
 javaChannel().socket().bind(localAddress, config.getBacklog());
 }
}

关键代码 io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive

public void channelActive(ChannelHandlerContext ctx) {
 ctx.fireChannelActive();
	// 触发 read (NioServerSocketChannel 上的 read 不是读取数据，只是为了触发 channel 的事件注册)
 readIfIsAutoRead();
}

关键代码 io.netty.channel.nio.AbstractNioChannel#doBeginRead

protected void doBeginRead() throws Exception {
 // Channel.read() or ChannelHandlerContext.read() was called
 final SelectionKey selectionKey = this.selectionKey;
 if (!selectionKey.isValid()) {
 return;
 }
 readPending = true;
 final int interestOps = selectionKey.interestOps();
 // readInterestOp 取值是 16，在 NioServerSocketChannel 创建时初始化好，代表关注 accept 事件
 if ((interestOps & readInterestOp) == 0) {
 selectionKey.interestOps(interestOps | readInterestOp);
 }
}

NioEventLoop 剖析

NioEventLoop 线程不仅要处理 IO 事件，还要处理 Task（包括普通任务和定时任务），

提交任务代码 io.netty.util.concurrent.SingleThreadEventExecutor#execute

public void execute(Runnable task) {
 if (task == null) {
 throw new NullPointerException("task");
 }
 boolean inEventLoop = inEventLoop();
 // 添加任务，其中队列使用了 jctools 提供的 mpsc 无锁队列
 addTask(task);
 if (!inEventLoop) {
 // inEventLoop 如果为 false 表示由其它线程来调用 execute，即首次调用，这时需要向 eventLoop 提交首个任务，启动死循环，会执行到下面的 doStartThread
 startThread();
 if (isShutdown()) {
 // 如果已经 shutdown，做拒绝逻辑，代码略...
 }
 }
 if (!addTaskWakesUp && wakesUpForTask(task)) {
 // 如果线程由于 IO select 阻塞了，添加的任务的线程需要负责唤醒 NioEventLoop 线程
 wakeup(inEventLoop);
 }
}

唤醒 select 阻塞线程io.netty.channel.nio.NioEventLoop#wakeup

@Override
protected void wakeup(boolean inEventLoop) {
 if (!inEventLoop && wakenUp.compareAndSet(false, true)) {
 selector.wakeup();
 }
}

启动 EventLoop 主循环 io.netty.util.concurrent.SingleThreadEventExecutor#doStartThread

private void doStartThread() {
 assert thread == null;
 executor.execute(new Runnable() {
 @Override
 public void run() {
 // 将线程池的当前线程保存在成员变量中，以便后续使用
 thread = Thread.currentThread();
 if (interrupted) {
 thread.interrupt();
 }
 boolean success = false;
 updateLastExecutionTime();
 try {
 // 调用外部类 SingleThreadEventExecutor 的 run 方法，进入死循环，run 方法见下
 SingleThreadEventExecutor.this.run();
 success = true;
 } catch (Throwable t) {
 logger.warn("Unexpected exception from an event executor: ", t);
 } finally {
	// 清理工作，代码略...
 }
 }
 });
}

io.netty.channel.nio.NioEventLoop#run 主要任务是执行死循环，不断看有没有新任务，有没有 IO 事件

protected void run() {
 for (;;) {
 try {
 try {
 // calculateStrategy 的逻辑如下：
 // 有任务，会执行一次 selectNow，清除上一次的 wakeup 结果，无论有没有 IO 事件，都会跳过 switch
 // 没有任务，会匹配 SelectStrategy.SELECT，看是否应当阻塞
 switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
 case SelectStrategy.CONTINUE:
 continue;
 case SelectStrategy.BUSY_WAIT:
 case SelectStrategy.SELECT:
 // 因为 IO 线程和提交任务线程都有可能执行 wakeup，而 wakeup 属于比较昂贵的操作，因此使用了一个原子布尔对象 wakenUp，它取值为 true 时，表示该由当前线程唤醒
 // 进行 select 阻塞，并设置唤醒状态为 false
 boolean oldWakenUp = wakenUp.getAndSet(false);
 
 // 如果在这个位置，非 EventLoop 线程抢先将 wakenUp 置为 true，并 wakeup
 // 下面的 select 方法不会阻塞
 // 等 runAllTasks 处理完成后，到再循环进来这个阶段新增的任务会不会及时执行呢?
 // 因为 oldWakenUp 为 true，因此下面的 select 方法就会阻塞，直到超时
 // 才能执行，让 select 方法无谓阻塞
 select(oldWakenUp);
 if (wakenUp.get()) {
 selector.wakeup();
 }
 default:
 }
 } catch (IOException e) {
 rebuildSelector0();
 handleLoopException(e);
 continue;
 }
 cancelledKeys = 0;
 needsToSelectAgain = false;
 // ioRatio 默认是 50
 final int ioRatio = this.ioRatio;
 if (ioRatio == 100) {
 try {
 processSelectedKeys();
 } finally {
 // ioRatio 为 100 时，总是运行完所有非 IO 任务
 runAllTasks();
 }
 } else { 
 final long ioStartTime = System.nanoTime();
 try {
 processSelectedKeys();
 } finally {
 // 记录 io 事件处理耗时
 final long ioTime = System.nanoTime() - ioStartTime;
 // 运行非 IO 任务，一旦超时会退出 runAllTasks
 runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
 }
 }
 } catch (Throwable t) {
 handleLoopException(t);
 }
 try {
 if (isShuttingDown()) {
 closeAll();
 if (confirmShutdown()) {
 return;
 }
 }
 } catch (Throwable t) {
 handleLoopException(t);
 }
 }
}

注意

这里有个费解的地方就是 wakeup，它既可以由提交任务的线程来调用（比较好理解），也可以由 EventLoop 线程来调用（比较费解），这里要知道 wakeup 方法的效果：
由非 EventLoop 线程调用，会唤醒当前在执行 select 阻塞的 EventLoop 线程
由 EventLoop 自己调用，会本次的 wakeup 会取消下一次的 select 操作

io.netty.channel.nio.NioEventLoop#select

private void select(boolean oldWakenUp) throws IOException {
 Selector selector = this.selector;
 try {
 int selectCnt = 0;
 long currentTimeNanos = System.nanoTime();
 // 计算等待时间
 // * 没有 scheduledTask，超时时间为 1s
 // * 有 scheduledTask，超时时间为 `下一个定时任务执行时间 - 当前时间`
 long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
 for (;;) {
 long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
 // 如果超时，退出循环
 if (timeoutMillis <= 0) {
 if (selectCnt == 0) {
 selector.selectNow();
 selectCnt = 1;
 }
 break;
 }
 // 如果期间又有 task 退出循环，如果没这个判断，那么任务就会等到下次 select 超时时才能被执行
 // wakenUp.compareAndSet(false, true) 是让非 NioEventLoop 不必再执行 wakeup
 if (hasTasks() && wakenUp.compareAndSet(false, true)) {
 selector.selectNow();
 selectCnt = 1;
 break;
 }
 // select 有限时阻塞
 // 注意 nio 有 bug，当 bug 出现时，select 方法即使没有时间发生，也不会阻塞住，导致不断空轮询，cpu 占用 100%
 int selectedKeys = selector.select(timeoutMillis);
 // 计数加 1
 selectCnt ++;
 // 醒来后，如果有 IO 事件、或是由非 EventLoop 线程唤醒，或者有任务，退出循环
 if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
 break;
 }
 if (Thread.interrupted()) {
 	// 线程被打断，退出循环
 // 记录日志
 selectCnt = 1;
 break;
 }
 long time = System.nanoTime();
 if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
 // 如果超时，计数重置为 1，下次循环就会 break
 selectCnt = 1;
 } 
 // 计数超过阈值，由 io.netty.selectorAutoRebuildThreshold 指定，默认 512
 // 这是为了解决 nio 空轮询 bug
 else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
 selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
 // 重建 selector
 selector = selectRebuildSelector(selectCnt);
 selectCnt = 1;
 break;
 }
 currentTimeNanos = time;
 }
 if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
 // 记录日志
 }
 } catch (CancelledKeyException e) {
 // 记录日志
 }
}

处理 keys io.netty.channel.nio.NioEventLoop#processSelectedKeys

private void processSelectedKeys() {
 if (selectedKeys != null) {
 // 通过反射将 Selector 实现类中的就绪事件集合替换为 SelectedSelectionKeySet 
 // SelectedSelectionKeySet 底层为数组实现，可以提高遍历性能（原本为 HashSet）
 processSelectedKeysOptimized();
 } else {
 processSelectedKeysPlain(selector.selectedKeys());
 }
}

io.netty.channel.nio.NioEventLoop#processSelectedKey

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
 final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
 // 当 key 取消或关闭时会导致这个 key 无效
 if (!k.isValid()) {
 // 无效时处理...
 return;
 }
 try {
 int readyOps = k.readyOps();
 // 连接事件
 if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
 int ops = k.interestOps();
 ops &= ~SelectionKey.OP_CONNECT;
 k.interestOps(ops);
 unsafe.finishConnect();
 }
 // 可写事件
 if ((readyOps & SelectionKey.OP_WRITE) != 0) {
 ch.unsafe().forceFlush();
 }
 // 可读或可接入事件
 if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
 // 如果是可接入 io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe#read
 // 如果是可读 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read
 unsafe.read();
 }
 } catch (CancelledKeyException ignored) {
 unsafe.close(unsafe.voidPromise());
 }
}

accept 剖析

nio 中如下代码，在 netty 中的流程

//1 阻塞直到事件发生
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) { 
 //2 拿到一个事件
 SelectionKey key = iter.next();
 
 //3 如果是 accept 事件
 if (key.isAcceptable()) {
 
 //4 执行 accept
 SocketChannel channel = serverSocketChannel.accept();
 channel.configureBlocking(false);
 
 //5 关注 read 事件
 channel.register(selector, SelectionKey.OP_READ);
 }
 // ...
}

先来看可接入事件处理（accept）

io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe#read

public void read() {
 assert eventLoop().inEventLoop();
 final ChannelConfig config = config();
 final ChannelPipeline pipeline = pipeline(); 
 final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
 allocHandle.reset(config);
 boolean closed = false;
 Throwable exception = null;
 try {
 try {
 do {
	// doReadMessages 中执行了 accept 并创建 NioSocketChannel 作为消息放入 readBuf
 // readBuf 是一个 ArrayList 用来缓存消息
 int localRead = doReadMessages(readBuf);
 if (localRead == 0) {
 break;
 }
 if (localRead < 0) {
 closed = true;
 break;
 }
	// localRead 为 1，就一条消息，即接收一个客户端连接
 allocHandle.incMessagesRead(localRead);
 } while (allocHandle.continueReading());
 } catch (Throwable t) {
 exception = t;
 }
 int size = readBuf.size();
 for (int i = 0; i < size; i ++) {
 readPending = false;
 // 触发 read 事件，让 pipeline 上的 handler 处理，这时是处理
 // io.netty.bootstrap.ServerBootstrap.ServerBootstrapAcceptor#channelRead
 pipeline.fireChannelRead(readBuf.get(i));
 }
 readBuf.clear();
 allocHandle.readComplete();
 pipeline.fireChannelReadComplete();
 if (exception != null) {
 closed = closeOnReadError(exception);
 pipeline.fireExceptionCaught(exception);
 }
 if (closed) {
 inputShutdown = true;
 if (isOpen()) {
 close(voidPromise());
 }
 }
 } finally {
 if (!readPending && !config.isAutoRead()) {
 removeReadOp();
 }
 }
}

关键代码 io.netty.bootstrap.ServerBootstrap.ServerBootstrapAcceptor#channelRead

public void channelRead(ChannelHandlerContext ctx, Object msg) {
 // 这时的 msg 是 NioSocketChannel
 final Channel child = (Channel) msg;
 // NioSocketChannel 添加 childHandler 即初始化器
 child.pipeline().addLast(childHandler);
 // 设置选项
 setChannelOptions(child, childOptions, logger);
 for (Entry<AttributeKey<?>, Object> e: childAttrs) {
 child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
 }
 try {
 // 注册 NioSocketChannel 到 nio worker 线程，接下来的处理也移交至 nio worker 线程
 childGroup.register(child).addListener(new ChannelFutureListener() {
 @Override
 public void operationComplete(ChannelFuture future) throws Exception {
 if (!future.isSuccess()) {
 forceClose(child, future.cause());
 }
 }
 });
 } catch (Throwable t) {
 forceClose(child, t);
 }
}

又回到了熟悉的 io.netty.channel.AbstractChannel.AbstractUnsafe#register 方法

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
 // 一些检查，略...
 AbstractChannel.this.eventLoop = eventLoop;
 if (eventLoop.inEventLoop()) {
 register0(promise);
 } else {
 try {
 // 这行代码完成的事实是 nio boss -> nio worker 线程的切换
 eventLoop.execute(new Runnable() {
 @Override
 public void run() {
 register0(promise);
 }
 });
 } catch (Throwable t) {
 // 日志记录...
 closeForcibly();
 closeFuture.setClosed();
 safeSetFailure(promise, t);
 }
 }
}

io.netty.channel.AbstractChannel.AbstractUnsafe#register0

private void register0(ChannelPromise promise) {
 try {
 if (!promise.setUncancellable() || !ensureOpen(promise)) {
 return;
 }
 boolean firstRegistration = neverRegistered;
 doRegister();
 neverRegistered = false;
 registered = true;
	
 // 执行初始化器，执行前 pipeline 中只有 head -> 初始化器 -> tail
 pipeline.invokeHandlerAddedIfNeeded();
 // 执行后就是 head -> logging handler -> my handler -> tail
 safeSetSuccess(promise);
 pipeline.fireChannelRegistered();
 
 if (isActive()) {
 if (firstRegistration) {
 // 触发 pipeline 上 active 事件
 pipeline.fireChannelActive();
 } else if (config().isAutoRead()) {
 beginRead();
 }
 }
 } catch (Throwable t) {
 closeForcibly();
 closeFuture.setClosed();
 safeSetFailure(promise, t);
 }
}

回到了熟悉的代码 io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive

public void channelActive(ChannelHandlerContext ctx) {
 ctx.fireChannelActive();
	// 触发 read (NioSocketChannel 这里 read，只是为了触发 channel 的事件注册，还未涉及数据读取)
 readIfIsAutoRead();
}

io.netty.channel.nio.AbstractNioChannel#doBeginRead

protected void doBeginRead() throws Exception {
 // Channel.read() or ChannelHandlerContext.read() was called
 final SelectionKey selectionKey = this.selectionKey;
 if (!selectionKey.isValid()) {
 return;
 }
 readPending = true;
	// 这时候 interestOps 是 0
 final int interestOps = selectionKey.interestOps();
 if ((interestOps & readInterestOp) == 0) {
 // 关注 read 事件
 selectionKey.interestOps(interestOps | readInterestOp);
 }
}

read 剖析

再来看可读事件 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read，注意发送的数据未必能够一次读完，因此会触发多次 nio read 事件，一次事件内会触发多次 pipeline read，一次事件会触发一次 pipeline read complete

public final void read() {
 final ChannelConfig config = config();
 if (shouldBreakReadReady(config)) {
 clearReadPending();
 return;
 }
 final ChannelPipeline pipeline = pipeline();
 // io.netty.allocator.type 决定 allocator 的实现
 final ByteBufAllocator allocator = config.getAllocator();
 // 用来分配 byteBuf，确定单次读取大小
 final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
 allocHandle.reset(config);
 ByteBuf byteBuf = null;
 boolean close = false;
 try {
 do {
 byteBuf = allocHandle.allocate(allocator);
 // 读取
 allocHandle.lastBytesRead(doReadBytes(byteBuf));
 if (allocHandle.lastBytesRead() <= 0) {
 byteBuf.release();
 byteBuf = null;
 close = allocHandle.lastBytesRead() < 0;
 if (close) {
 readPending = false;
 }
 break;
 }
 allocHandle.incMessagesRead(1);
 readPending = false;
 // 触发 read 事件，让 pipeline 上的 handler 处理，这时是处理 NioSocketChannel 上的 handler
 pipeline.fireChannelRead(byteBuf);
 byteBuf = null;
 } 
 // 是否要继续循环
 while (allocHandle.continueReading());
 allocHandle.readComplete();
 // 触发 read complete 事件
 pipeline.fireChannelReadComplete();
 if (close) {
 closeOnRead(pipeline);
 }
 } catch (Throwable t) {
 handleReadException(pipeline, byteBuf, t, close, allocHandle);
 } finally {
 if (!readPending && !config.isAutoRead()) {
 removeReadOp();
 }
 }
}

io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator.MaxMessageHandle#continueReading(io.netty.util.UncheckedBooleanSupplier)

public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
 return 
 // 一般为 true
 config.isAutoRead() &&
 // respectMaybeMoreData 默认为 true
 // maybeMoreDataSupplier 的逻辑是如果预期读取字节与实际读取字节相等，返回 true
 (!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
 // 小于最大次数，maxMessagePerRead 默认 16
 totalMessages < maxMessagePerRead &&
 // 实际读到了数据
 totalBytesRead > 0;
}

面试题专栏

Java面试题专栏已上线，欢迎访问。

如果你不知道简历怎么写，简历项目不知道怎么包装；
如果简历中有些内容你不知道该不该写上去；
如果有些综合性问题你不知道怎么答；

那么可以私信我，我会尽我所能帮助你。

Netty高级使用与源码详解

粘包与半包

粘包现象

半包现象

现象分析

解决方案

方法1：短链接（极不推荐）

方法2：固定长度

方法3：固定分隔符

方法4：预设长度

协议设计与解析

为什么需要协议？

redis 协议举例

http 协议举例

自定义协议要素

编解码器

@Sharable

扩展序列化算法

参数调优

客户端参数 CONNECT_TIMEOUT_MILLIS

服务端参数 SO_BACKLOG

TCP_NODELAY

SO_SNDBUF & SO_RCVBUF

ALLOCATOR

RCVBUF_ALLOCATOR

源码详解

启动剖析

NioEventLoop 剖析

注意

accept 剖析

read 剖析

面试题专栏

%s 个评论

最新文章