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 c.i.n.HelloWorldServer - binding...08:24:46 c.i.n.HelloWorldServer - bound...08:24:55 i.n.h.l.LoggingHandler - REGISTERED08:24:55 i.n.h.l.LoggingHandler - ACTIVE08:24:55 c.i.n.HelloWorldServer - connected 08:24:55 i.n.h.l.LoggingHandler - READ: 160B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - 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 c.i.n.HelloWorldServer - binding...08:43:49 c.i.n.HelloWorldServer - bound...08:44:23 i.n.h.l.LoggingHandler - REGISTERED08:44:23 i.n.h.l.LoggingHandler - ACTIVE08:44:23 c.i.n.HelloWorldServer - connected 08:44:24 i.n.h.l.LoggingHandler - READ: 20B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - READ COMPLETE08:44:24 i.n.h.l.LoggingHandler - READ: 140B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - 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; for (int j = 0; j < r.nextInt(8); j++) { bytes = (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 c.i.n.HelloWorldClient - connetted...12:07:00 i.n.h.l.LoggingHandler - REGISTERED12:07:00 i.n.h.l.LoggingHandler - CONNECT: /192.168.0.103:909012:07:00 i.n.h.l.LoggingHandler - ACTIVE12:07:00 c.i.n.HelloWorldClient - sending...12:07:00 i.n.h.l.LoggingHandler - WRITE: 80B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - FLUSH服务端输出
12:06:51 c.i.n.HelloWorldServer - binding...12:06:51 c.i.n.HelloWorldServer - bound...12:07:00 i.n.h.l.LoggingHandler - REGISTERED12:07:00 i.n.h.l.LoggingHandler - ACTIVE12:07:00 c.i.n.HelloWorldServer - connected 12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 61 61 61 61 00 00 00 00 |aaaa.... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 62 00 00 00 00 00 00 00 |b....... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 63 63 00 00 00 00 00 00 |cc...... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 64 00 00 00 00 00 00 00 |d....... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 00 00 00 00 00 00 00 00 |........ |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 66 66 66 66 00 00 00 00 |ffff.... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 67 67 67 00 00 00 00 00 |ggg..... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 68 00 00 00 00 00 00 00 |h....... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 69 69 69 69 69 00 00 00 |iiiii... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 6a 6a 6a 6a 00 00 00 00 |jjjj.... |+--------+-------------------------------------------------+----------------+12:07:00 i.n.h.l.LoggingHandler - 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 c.i.n.HelloWorldClient - connetted...14:08:18 i.n.h.l.LoggingHandler - REGISTERED14:08:18 i.n.h.l.LoggingHandler - CONNECT: /192.168.0.103:909014:08:18 i.n.h.l.LoggingHandler - ACTIVE14:08:18 c.i.n.HelloWorldClient - sending...14:08:18 i.n.h.l.LoggingHandler - WRITE: 60B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - FLUSH服务端输出
14:08:18 c.i.n.HelloWorldServer - connected 14:08:18 i.n.h.l.LoggingHandler - READ: 1B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 61 |a |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 3B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 62 62 62 |bbb |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 3B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 63 63 63 |ccc |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 2B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 64 64 |dd |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 10B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeee |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 2B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 66 66 |ff |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 7B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 67 67 67 67 67 67 67 |ggggggg |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 4B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 68 68 68 68 |hhhh |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 7B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 69 69 69 69 69 69 69 |iiiiiii |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - READ: 11B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjjjj |+--------+-------------------------------------------------+----------------+14:08:18 i.n.h.l.LoggingHandler - 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 c.i.n.HelloWorldClient - connetted...14:37:10 i.n.h.l.LoggingHandler - REGISTERED14:37:10 i.n.h.l.LoggingHandler - CONNECT: /192.168.0.103:909014:37:10 i.n.h.l.LoggingHandler - ACTIVE14:37:10 c.i.n.HelloWorldClient - sending...14:37:10 i.n.h.l.LoggingHandler - WRITE: 97B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|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 i.n.h.l.LoggingHandler - FLUSH服务端输出
14:36:50 c.i.n.HelloWorldServer - binding...14:36:51 c.i.n.HelloWorldServer - bound...14:37:10 i.n.h.l.LoggingHandler - REGISTERED14:37:10 i.n.h.l.LoggingHandler - ACTIVE14:37:10 c.i.n.HelloWorldServer - connected 14:37:10 i.n.h.l.LoggingHandler - READ: 9B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 61 61 61 61 61 61 61 61 61 |aaaaaaaaa |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 9B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 62 62 62 62 62 62 62 62 62 |bbbbbbbbb |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 6B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 63 63 63 63 63 63 |cccccc |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 8B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 64 64 64 64 64 64 64 64 |dddddddd |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 15B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 65 65 65 65 65 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeeeeeeee |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 13B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 66 66 66 66 66 66 66 66 66 66 66 66 66 |fffffffffffff |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 2B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 67 67 |gg |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 2B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 68 68 |hh |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 14B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 69 69 69 69 69 69 69 69 69 69 69 69 69 69 |iiiiiiiiiiiiii|+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - READ: 9B +-------------------------------------------------+ |0123456789abcdef |+--------+-------------------------------------------------+----------------+|00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjj |+--------+-------------------------------------------------+----------------+14:37:10 i.n.h.l.LoggingHandler - 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 完成收发
@Slf4jpublic 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; 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());// encodeLoginRequestMessage message = new LoginRequestMessage("zhangsan", "123", "张三");// channel.writeOutbound(message);// decodeByteBuf 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(); // 引用计数 2channel.writeInbound(s1); // release 1channel.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; 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;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; }}增加配置类和配置文件
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; in.readBytes(bytes, 0, length); // 找到反序列化算法 Serializer.Algorithm algorithm = Serializer.Algorithm.values(); // 确定具体消息类型 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
@Datapublic 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 等都是无限等待的，如果不希望永远阻塞，使用它调整超时时间
@Slf4jpublic 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
@Overridepublic 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 线程）来封装线程和 selectorSelector selector = Selector.open(); //2 创建 NioServerSocketChannel，同时会初始化它关联的 handler，以及为原生 ssc 存储 configNioServerSocketChannel attachment = new NioServerSocketChannel();//3 创建 NioServerSocketChannel 时，创建了 java 原生的 ServerSocketChannelServerSocketChannel 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 处理
[*]init：由main处理
[*]创建NioServerSocketChannel：由main处理
[*]添加 NioServerSocketChannel 初始化 handler ：由main处理
[*]初始化 handler 等待调用
[*]register
[*]启动 nio boss 线程 ：由main处理
[*]原生 ssc 注册至 selector 未关注事件：由nio-thread处理
[*]执行 NioServerSocketChannel 初始化 handler：由nio-thread处理
[*]regFuture 等待回调 doBind0：由nio-thread处理
[*]原生 ServerSocketChannel 绑定：由nio-thread处理
[*]触发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 实际上是 NioServerSocketChannelvoid 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 执行 doBind0private 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
@Overrideprotected 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面试题专栏已上线，欢迎访问。
[*]如果你不知道简历怎么写，简历项目不知道怎么包装；
[*]如果简历中有些内容你不知道该不该写上去；
[*]如果有些综合性问题你不知道怎么答；
那么可以私信我，我会尽我所能帮助你。
页:
[1]