Netty源码分析之IO事件
IO事件概述
在上节我们知道Netty启动后会动起一个selector线程监听IO事件，IO事件包括以下几个：
- SelectionKey.OP_READ 读操作位触发读事件
- SelectionKey.OP_WRITE 写操作位触发写事件
- SelectionKey.OP_ACCEPT 接收操作位触发接收事件
- SelectionKey.OP_CONNECT 连接操作位触发连接事件
读事件
读事件即可以发生在客户端也可会发生在服务端，当客户端或服务端注册读事件并接受到远端发送的数据就会触发读事件。
写事件
写事件即可以发生在客户端也可会发生在服务端，写事件可以由外部直接调用触发，当出现写半包时（出现在TCP缓存满的情况），Netty会注册写操作位，待TCP缓存消耗后也会触发写事件。
接收事件
只发生在服务端，服务端启动的时候会注册接收操作位监听客户端的连接。
连接事件
只发生在客户端，客户端启动时会尝试连接服务端，连接是异步的不一定马上成功不成功则需要注册连接操作位监听客户端的连接成功。
下面从服务端的角度介绍Netty启动后，接收客户端连接的流程，以及客户端连接上后服务端的读和写的流程。
接收事件流程
当Selector轮询到接收事件会在NioEventLoop类中的processSelectedKey方法中进行处理，源码如下：
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
...
int readyOps = k.readyOps();
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
if (!ch.isOpen()) {
return;
}
}
...
}
服务端的接收逻辑直接委托给unsafe.read()处理，unsafe中有2个实现类NioMessageUnsafe和NioByteUnsafe，由于服务端启动初始化的Channel用的是NioServerSocketChannel，所以unsafe的实现类是NioMessageUnsafe，下面看下unsafe.read()的实现：
public void read() {
...
try {
for (;;) {
//获取接收结果
int localRead = doReadMessages(readBuf);
//如果接收结果为空直接推出
if (localRead == 0) {
break;
}
//异常情况下返回，tcp协议未用到
if (localRead < 0) {
closed = true;
break;
}
//非自动读，退出并去掉监听客户端的连接事件，变成手工注册，一般不用
if (!config.isAutoRead()) {
break;
}
//每波的最大处理连接请求数默认为16
if (readBuf.size() >= maxMessagesPerRead) {
break;
}
}
} catch (Throwable t) {
exception = t;
}
setReadPending(false);
int size = readBuf.size();
//调用pipeline链处理客户端连接事件
for (int i = 0; i < size; i ++) {
pipeline.fireChannelRead(readBuf.get(i));
}
//清理接收对象
readBuf.clear();
//调用pipeline链处理接收完成事件
pipeline.fireChannelReadComplete();
...
}
接收客户端的处理逻辑主要流程如下：
- 调用doReadMessages方法获取客户端连接，doReadMessages的实现其实就是调用了JDK原生的accept()方法获取与客户端通讯的通道SocketChannel（系统包装成NioSocketChannel类）。
- 获取到SocketChannel后交给pipeline.fireChannelRead方法做进一步处理。
SocketChannel是服务端和客户端通讯的核心操作类，pipeline.fireChannelRead方法在之前讲过是一个调用链，调用用户的配置ChannelHandler，这里系统会调用初始化channel时系统自动注册的ServerBootstrapAcceptor里的channelRead方法（初始化channel流程可以阅读上节内容）
channelRead是Netty的核心代码主要对SocketChannel进一步封装使剥离AcceptorSelector线程，独立出跟客户端通讯IOSelector线程。我们来看下它的实现：
public void channelRead(ChannelHandlerContext ctx, Object msg) {
//获取与客户端通讯的通道SocketChannel（下面叫childChannel）
final Channel child = (Channel) msg;
//将用户配置的处理器childChannel设置到childChannel
child.pipeline().addLast(childHandler);
//将用户配置的系统参数设置到childChannel
for (Entry<ChannelOption<?>, Object> e: childOptions) {
try {
if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
logger.warn("Unknown channel option: " + e);
}
} catch (Throwable t) {
logger.warn("Failed to set a channel option: " + child, t);
}
}
//将用户配置的属性设置到childChannel
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
//在从线程池里注册childChannel
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);
}
}
channelRead流程比较简单就是在从线程池注册childChannel，然后从线程池起相应的selector线程处理服务端和客户端的读事件和写事件。
这里childChannel的注册流程和服务端启动时的channel的注册流程基本一样不过这里默认自动注册SelectionKey.OP_READ读事件而不是SelectionKey.OP_ACCEPT接收事件。需要注意的是从线程池的每个线程会创建一个selector对象而一个selector可能注册多个childChannel。
读事件流程
完成上面的流程客户端就可以跟服务端通讯了，如果客户端发送了数据，服务端的从selector线程就会了轮询到读事件，同样读事件会在NioEventLoop类中的processSelectedKey方法中进行处理，源码如下：
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
...
int readyOps = k.readyOps();
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
if (!ch.isOpen()) {
return;
}
}
...
}
可以看到读事件源码跟接收事件的源码是同一块，只是unsafe实现的类不一样，由于childChannel的实现类是NioSocketChannel因此unsafe的实现类是NioByteUnsafe，下面重点分析下unsafe.read()的实现：
@Override
public final void read() {
final ChannelConfig config = config();
//readPending状态是非自动读情况下使用，readPending如果是false表示数据已读完移除读操作位
if (!config.isAutoRead() && !isReadPending()) {
removeReadOp();
return;
}
final ChannelPipeline pipeline = pipeline();
//获取ByteBuf构造器
final ByteBufAllocator allocator = config.getAllocator();
//获取自动读模式下的一次性读取的最大的次数
final int maxMessagesPerRead = config.getMaxMessagesPerRead();
//获取ByteBuf容量分配器
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
ByteBuf byteBuf = null;
int messages = 0;
boolean close = false;
try {
int totalReadAmount = 0;
boolean readPendingReset = false;
do {
//构造ByteBuf
byteBuf = allocHandle.allocate(allocator);
//获取byteBuf最大可写字节
int writable = byteBuf.writableBytes();
//将接收到的字段写入到byteBuf，并获取接收数据长度
int localReadAmount = doReadBytes(byteBuf);
//未读到数据释放byteBuf，跳出读取逻辑
if (localReadAmount <= 0) {
byteBuf.release();
byteBuf = null;
close = localReadAmount < 0;
break;
}
//数据读完设置readPending为false
if (!readPendingReset) {
readPendingReset = true;
setReadPending(false);
}
//交给用户配置的数据解析器（ChannelHandler）处理读到的数据
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
//避免溢出，如果读取的数据量大于Integer的最大值则直接退出
if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
totalReadAmount = Integer.MAX_VALUE;
break;
}
//加上这次读到的数据量 计算出总数据量
totalReadAmount += localReadAmount;
//非自动读，退出
if (!config.isAutoRead()) {
break;
}
//如果读到的数据量小于byteBuf最大可写字节 说明数据已经接受完，退出循环
if (localReadAmount < writable) {
break;
}
//否则数据还未读完，继续读，直到读完或者读取次数大于最大次数
} while (++ messages < maxMessagesPerRead);
//触发读取完成的处理器
pipeline.fireChannelReadComplete();
//记录这次读取数据的总量，以便后续动态创建byteBuf的大小
allocHandle.record(totalReadAmount);
if (close) {
closeOnRead(pipeline);
close = false;
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close);
} finally {
if (!config.isAutoRead() && !isReadPending()) {
removeReadOp();
}
}
}
}
unsafe.read()的实现比较复杂我们按步骤来分析
- 先构造ByteBuf，构造ByteBuf用到了2个工具allocator和allocHandle。
allocator作用是构造怎么样类型的ByteBuf比如默认就是构造了UnpooledUnsafeDirectByteBuf，UnpooledUnsafeDirectByteBuf提供了非池化的堆外内存直接操作的支持。
allocHandle作用是分配ByteBuf的容量，allocHandle有2个实现类分别是FixedRecvByteBufAllocator和AdaptiveRecvByteBufAllocator，FixedRecvByteBufAllocator实现比较简单,根据用户的配置分配固定的容量，AdaptiveRecvByteBufAllocator是默认的实现，它会根据上次分配的容量动态调整大小。
static final int DEFAULT_MINIMUM = 64;
static final int DEFAULT_INITIAL = 1024;
static final int DEFAULT_MAXIMUM = 65536;
private static final int INDEX_INCREMENT = 4;
private static final int INDEX_DECREMENT = 1;
AdaptiveRecvByteBufAllocator的最小容量为64字节，默认初始容量为1024，最大容量为65536字节，其扩展步伐值为4，收缩步伐值为1。
static {
List<Integer> sizeTable = new ArrayList<Integer>();
for (int i = 16; i < 512; i += 16) {
sizeTable.add(i);
}
for (int i = 512; i > 0; i <<= 1) {
sizeTable.add(i);
}
SIZE_TABLE = new int[sizeTable.size()];
for (int i = 0; i < SIZE_TABLE.length; i ++) {
SIZE_TABLE[i] = sizeTable.get(i);
}
}
AdaptiveRecvByteBufAllocator初始化了53个容量选项当容量小于512字节时以16字节的步伐增长，当容量大于512字节时，容量以上一个容量的2倍增长。
我们重点分析下扩容策略方法record()
public void record(int actualReadBytes) {
//本次读取的总容量与上次容量收缩后的前一个位置的容量比较，如果比它还小或相等说明还在收缩
if (actualReadBytes <= SIZE_TABLE[Math.max(0, index - INDEX_DECREMENT - 1)]) {
//是否持续在收缩，是的话上次容量收缩1个步伐
if (decreaseNow) {
index = Math.max(index - INDEX_DECREMENT, minIndex);
nextReceiveBufferSize = SIZE_TABLE[index];
decreaseNow = false;
//不是持续收缩，标记下，不做收缩处理
} else {
decreaseNow = true;
}
//如果当前容量大于上次的容量，则扩容4个步伐
} else if (actualReadBytes >= nextReceiveBufferSize) {
index = Math.min(index + INDEX_INCREMENT, maxIndex);
nextReceiveBufferSize = SIZE_TABLE[index];
decreaseNow = false;
}
}
record()的参数actualReadBytes表示本次读取的总容量。以上代码的总结如下：
容量的收缩以下2个条件
- 1.如果本次读取的总容量比上次容量收缩后的前一个位置的容量还小或相等。
- 2.容量至少持续2次在收缩。
容量的扩展以下1个条件
- 1.如果本次读取的总容量大于上次的容量。
以上分析了构造ByteBuf，下面继续分析unsafe.read()方法。
-
将接收到的字段写入到byteBuf，并获取接收数据长度。
-
如果未读到数据，说明数据已经读物，释放byteBuf缓存，跳出读取循环逻辑。
这里的doReadBytes可能返回0，大于0和-1，等于0说明已无数据可读，大于0表示读到数据，-1表示读取异常需要设置close标志为true用于关闭连接。
-
避免溢出，如果读取的数据量大于Integer的最大值则直接跳出读取循环逻辑。
-
交给用户配置的数据解析器（ChannelHandler）处理读到的数据。
这里交给数据解析器数据，不一定是你想要的完整的数据，可能出现半包和粘包的现象，这需要数据解析器处理才能获得完整数据。
-
累计数据总量totalReadAmount，以便后面扩容策略用。
-
非自动读，跳出读取循环逻辑。
非自动情况下需要编程人员自己注册读操作位，才能触发读事件
-
如果读到的数据量小于byteBuf最大可写字节 说明数据已经接受完，跳出读取循环逻辑。
-
如果读到的数据量等于byteBuf最大可写字节，说明TCP缓存区还能还有数据，需要再次循环去读。
这里最大的循环次数模式是16次（可配）如果超过这个次数无论TCP缓存区是否还要数据都会终止循环，等下个selector周期再去读。
-
触发读取完成的处理器。
这里的处理器需要编程人员自己配置。
-
记录这次读取数据的总量，以便后续动态创建byteBuf的大小，动态扩容上面已经讲过，这里不累述。
-
如果读取发生IO异常，则关闭连接。(上面讲过的读取状态返回-1并且close标志设置为true的情况)。
以上就是整个读事件的整个流程，下面来分析下写事件流程。
写事件流程
服务端一般在接受到数据处理结束后给客户端端返回一个响应数据，发送响应数据则需要调用ChannelHandlerContext#writeAndFlush方法。我们以外部调用ChannelHandlerContext#writeAndFlush（以下称ctx.writeAndFlush）方法为例来分析写事件流程，ctx.writeAndFlush方法看字面意思就是写入和刷新，ctx.writeAndFlush的写入是不会真正的发送，而是存到缓存中，刷新后才从缓存拿出数据发送。ctx.writeAndFlush也是调用链，开发人员可以实现ChannelOutboundHandler里的
void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;
方法捕获写数据事件或者实现
void flush(ChannelHandlerContext ctx) throws Exception;
方法来捕获刷新数据事件。
ctx.writeAndFlush的真正实现是其私有方法write，我看来看下write的源码实现：
private void write(Object msg, boolean flush, ChannelPromise promise) {
//获取下个ChannelOutboundHandler的包装ChannelHandlerContext
AbstractChannelHandlerContext next = findContextOutbound();
//获取ChannelHandlerContext里分配的线程
EventExecutor executor = next.executor();
//如果是跟当前线程同一个
if (executor.inEventLoop()) {
//直接调用write
next.invokeWrite(msg, promise);
//如果需要刷新调用flush
if (flush) {
next.invokeFlush();
}
//如果非同一个线程，需要异步处理
} else {
//获取可读取数据量
int size = channel.estimatorHandle().size(msg);
if (size > 0) {
ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer();
//因为异步发送事先增加待发送缓存量，占用空间
if (buffer != null) {
buffer.incrementPendingOutboundBytes(size);
}
}
Runnable task;
//如果需要刷新，创建带刷新的任务
if (flush) {
task = WriteAndFlushTask.newInstance(next, msg, size, promise);
//否则创建不带刷新的任务
} else {
task = WriteTask.newInstance(next, msg, size, promise);
}
safeExecute(executor, task, promise, msg);
}
}
实现比较简单，取链中下个ChannelHandlerContext，这里的ChannelHandlerContext是ChannelHandler的包装类，维护了ChannelHandlerContext的next和prev节点。
如果ChannelHandlerContext中配置的线程跟当前是同线程则同步调用写和刷新的方法，这里如果ChannelHandlerContext中没有配置线程默认取的channel里的线程。
如果不是当前线程就要创建任务异步执行写和刷新的方法，这里如果是异步执行的话会事先增加待发送缓存量预占用空间，待要执行写的方法的时候会还原缓存占用空间。
增加待发送缓存量主要是为了反正发送的数据太大导致缓存消耗速度小于写入速度，如果超过用户配置的大小（默认64k），会给用户配置的处理器发报警，这里涉及Netty缓存设计，不过多介绍。
netty写的过程会调用用户配置的处理器，这里编程人员可以实现ChannelOutboundHandler的
void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;
方法来捕获写事件（比如对数据进行编码）。当然写的最终流程是在channel的unsafe里执行：
public final void write(Object msg, ChannelPromise promise) {
...
int size;
try {
//包装msg
msg = filterOutboundMessage(msg);
//获取数据大小
size = estimatorHandle().size(msg);
if (size < 0) {
size = 0;
}
} catch (Throwable t) {
safeSetFailure(promise, t);
ReferenceCountUtil.release(msg);
return;
}
//写入缓存
outboundBuffer.addMessage(msg, size, promise);
}
写入缓存前，会对msg进行包装，这里的包装主要将存储在Java堆内存的待数据写入到堆外内存，存在堆外内存好处就是真正的发送时减少一次堆内拷贝到堆外的过程，提升发送效率。
最后将转化好的msg写入到outboundBuffer缓存，这里的写入也会执行一次增加待发送缓存量操作，所以上面讲的在异步写入的写入操作真正执行前会会还原缓存占用空间，为的就是避免重复的增加待发送缓存量操作。
刷新的过程成也是一样，会先调用用户配置的处理器，这里编程人员可以实现ChannelOutboundHandler的
void flush(ChannelHandlerContext ctx) throws Exception;
方法来捕获刷新事件，当然写的最终流程也是在channel的unsafe里执行：
public final void flush() {
//获取缓存容器
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
//如果缓存已经释放，则退出
if (outboundBuffer == null) {
return;
}
//准备待发送数据
outboundBuffer.addFlush();
//刷新
flush0();
}
比较简单，先获取缓存然后准备待发送数据最后调用flush0()刷新，这里需要注意的是发送相关的处理不要配置成多线程，这里会出现并发准备待发送数据的问题。
下面来看下flush0()的实现：
protected void flush0() {
//避免再次进入
if (inFlush0) {
return;
}
//获取缓存容器
final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null || outboundBuffer.isEmpty()) {
return;
}
inFlush0 = true;
//如果当前通道（channel）已经关闭，或断开连接，则执行删除当前待发送数据操作。
if (!isActive()) {
try {
if (isOpen()) {
outboundBuffer.failFlushed(NOT_YET_CONNECTED_EXCEPTION, true);
} else {
outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION, false);
}
} finally {
inFlush0 = false;
}
return;
}
try {
//发送数据操作
doWrite(outboundBuffer);
} catch (Throwable t) {
//出现IO异常并且配置自动关闭则关闭所有
if (t instanceof IOException && config().isAutoClose()) { close(voidPromise(), t, false);
//否则执行删除当前待发送数据操作。
} else {
outboundBuffer.failFlushed(t, true);
}
} finally {
inFlush0 = false;
}
}
flush0()主要有2个操作
- 如果当前通道（channel）已经关闭，或断开连接，则执行删除当前待发送数据操作，并退出。
- 否则调用doWrite方法发送数据。
我们重点分析下doWrite方法：
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
for (;;) {
//获取需要待发送数据个数
int size = in.size();
//如果待发送数据为空，清理写操作位并退出
if (size == 0) {
clearOpWrite();
break;
}
long writtenBytes = 0;
boolean done = false;
boolean setOpWrite = false;
//获取待发送数据
ByteBuffer[] nioBuffers = in.nioBuffers();
//获取待发送数据总个数
int nioBufferCnt = in.nioBufferCount();
//获取待发送数据总字节数
long expectedWrittenBytes = in.nioBufferSize();
//获取JDK的SocketChannel
SocketChannel ch = javaChannel();
switch (nioBufferCnt) {
//如果为0可能除了ByteBuffers类型外还要其他类型要发送，交给父类处理
case 0:
super.doWrite(in);
return;
//如果是单个待发送数据，调用JDK的SocketChannel单个发送方法
case 1:
ByteBuffer nioBuffer = nioBuffers[0];
for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
final int localWrittenBytes = ch.write(nioBuffer);
if (localWrittenBytes == 0) {
setOpWrite = true;
break;
}
expectedWrittenBytes -= localWrittenBytes;
writtenBytes += localWrittenBytes;
if (expectedWrittenBytes == 0) {
done = true;
break;
}
}
break;
//如果是多个待发送数据，调用JDK的SocketChannel多个发送方法
default:
//发送尝试，默认尝试16次
for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
//调用JDK的SocketChannel的write发送数据
final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
//如果发送的量为0，通道不可写，TCP缓存队列已满，设置写操作位标记setOpWrite为true，退出发送
if (localWrittenBytes == 0) {
setOpWrite = true;
break;
}
//期望发送字节数减掉已经成功发送字节数
expectedWrittenBytes -= localWrittenBytes;
//累加已发送字节数
writtenBytes += localWrittenBytes;
//期望发送字节数为0，说明已经发送完毕，设置发送结束标志done为true，退出发送
if (expectedWrittenBytes == 0) {
done = true;
break;
}
}
break;
}
// 释放完全写入的缓冲区，并更新部分写入的缓冲区的索引。
in.removeBytes(writtenBytes);
//如果数据未发送完，处理未发送完成逻辑
if (!done) {
incompleteWrite(setOpWrite);
break;
}
}
}
doWrite方法是一个大循环，它每次处理完会再试，直到没有需要待发送数据，就是删除写操作位（如果有）。
doWrite发送数据前先获取3个局部变量：
-
待发送数据数组nioBuffers。
-
获取待发送数据总个数nioBufferCnt。
-
获取期待待发送数据总字节数expectedWrittenBytes。
根据待发送数据总个数nioBufferCnt，发送要分3种情况。
-
如果nioBufferCnt为0，说明待发送的ByteBuffers类型数据为0，但可能除了ByteBuffers类型外还要其他类型要发送，交给父类处理。
-
如果nioBufferCnt为1，说明待发送的ByteBuffer数据只有1个，调用JDK的SocketChannel的实现单个数据发送的write方法。
-
如果nioBufferCnt大于1，说明待发送的ByteBuffer数据有多个，调用JDK的SocketChannel的实现多个数据发送的write方法。
第1种一般不会出现我们不做分析，重点分析下第2第3种情况，第2第3种方法除了调用JDK的SocketChannel的write方法实现不一样，其他逻辑完全相同。
发送的逻辑可能会出现一次发送不完全的情况这里默认尝试16次发送（可配），最终会出现3种最终发送情况：
-
发送数据完成。
-
发送完全失败，出现通道不可写情况。
-
尝试16次发送后只是部分发送成功，出现通道不可写或发送数据量太大情况。
第1和第3种情况是有数据发送成功的情况，所以发送完后会释放完全发送成功的缓冲区，并更新部分发送成功的缓冲区的索引。
第2和第3情况是有产生数据未发送成功的情况，所以会调用incompleteWrite做相应的后续处理。
我们来看下incompleteWrite方法实现：
protected final void incompleteWrite(boolean setOpWrite) {
// 如果出现通道不可写情况，则注册写操作位由selector异步轮询到OP_WRITE事件的时候调用foreceFlush进行flush
if (setOpWrite) {
setOpWrite();
//如果出现数据量太大情况，放入channel线程中排队处理未发送数据，以便在此期间可以执行其他任务
} else {
Runnable flushTask = this.flushTask;
if (flushTask == null) {
flushTask = this.flushTask = new Runnable() {
@Override
public void run() {
flush();
}
};
}
eventLoop().execute(flushTask);
}
}
incompleteWrite方法有2种情况：
-
如果出现通道不可写情况，则注册写操作位由selector异步轮询到OP_WRITE事件的时候去刷新。
出现不可写情况通常都是TCP缓存空间满了，只有TCP缓存预留空间大于发送低潮限度时才会触发OP_WRITE事件，所以出现通道不可写情况必须注册写操作位交给系统来判断TCP缓存空间是否可写。
-
如果出现数据量太大情况，放入channel线程中排队去刷新未发送数据，这样以便在此期间可以执行其他任务。
以上就是对Netty的IO事件的分析
