Netty原始碼分析之NioEventLoop(二)—NioEventLoop的啟動
摘要：
上篇文章中我們對Netty中NioEventLoop建立流程與原始碼進行了跟蹤分析。本篇文章中我們接著分析NioEventLoop的啟動流程；
Netty中會在服務端啟動和新連線接入時通過chooser選擇器，分別為NioServerSocketChannel與NioSocketChann...
上篇文章中我們對Netty中NioEventLoop建立流程與原始碼進行了跟蹤分析。本篇文章中我們接著分析NioEventLoop的啟動流程；
Netty中會在服務端啟動和新連線接入時通過chooser選擇器，分別為NioServerSocketChannel與NioSocketChannel選擇繫結一個NioEventLoop，接下來我們就分別從這兩個方面梳理NioEventLoop的啟動原始碼
一、服務端啟動
首先我們結合下圖看下Netty服務啟動過程中，NioServerSocketChannel繫結的NioEventLoop啟動流程
bind()部分原始碼我們在之前服務端啟動過程中進行過說明，我們進一步跟蹤進入doBind0()方法中可以看到channel.eventLoop().execute的執行，需要說明的是這裡其實啟動的NioServerSocketChannel繫結的 bossGroup，用來負責處理新連線接入的。
/**
* read by jsf
*
* @param regFuture
* @param channel
* @param localAddress
* @param promise
*/
private static void doBind0(final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress,
final ChannelPromise promise) {
//該方法向 NioServerSocketChannel 的 eventLoop 提交了一個任務，當 future(其實就是 promise) 成功後執行
//NioServerSocketChannel 的 bind 方法，並新增一個關閉監聽器。我們主要關注 bind 方法。
// This method is invoked before channelRegistered() is triggered. Give user
// handlers a chance to set up
// the pipeline in its channelRegistered() implementation.
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());
}
}
});
}
進入NioEventLoop父類SingleThreadEventExecutor中的execute方法，改方法通過inEventLoop()會首先判斷當前的執行緒是否是NioEventLoop本身繫結的執行緒，結合inEventLoop的程式碼可以看到NioEventLoop本身執行緒還未初始化為空，這裡返回false，執行啟動執行緒操作，同時會任務放入任務佇列中。
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
//首先判斷當前執行緒是否是該EventLoop繫結的執行緒
boolean inEventLoop = inEventLoop();
//把傳入的任務加入任務對立
addTask(task);
if (!inEventLoop) {//如果不是同一條執行緒
startThread();
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
@Override
public boolean inEventLoop(Thread thread) {
return thread == this.thread;
}
繼續跟蹤進入startThread()方法中
private void startThread() {
if (state == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
try {
doStartThread();
} catch (Throwable cause) {
STATE_UPDATER.set(this, ST_NOT_STARTED);
PlatformDependent.throwException(cause);
}
}
}
}
在 doStartThread()中主要實現了以下功能：
1、執行傳入的ThreadPerTaskExecutor的execute方法，建立一個新的執行緒，並與這個NioEventLoop物件繫結；
2、在開啟的執行緒中執行SingleThreadEventExecutor.this.run()，也就是NioEventLoop的run方法，開始NioEventLoop的執行操作；
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 {
//執行NioEventLoop的run方法
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
for (;;) {
int oldState = state;
if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
break;
}
}
// Check if confirmShutdown() was called at the end of the loop.
if (success && gracefulShutdownStartTime == 0) {
if (logger.isErrorEnabled()) {
logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must " +
"be called before run() implementation terminates.");
}
}
try {
// Run all remaining tasks and shutdown hooks.
for (;;) {
if (confirmShutdown()) {
break;
}
}
} finally {
try {
cleanup();
} finally {
STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
threadLock.release();
if (!taskQueue.isEmpty()) {
if (logger.isWarnEnabled()) {
logger.warn("An event executor terminated with " +
"non-empty task queue (" + taskQueue.size() + ')');
}
}
terminationFuture.setSuccess(null);
}
}
}
}
});
}
OK到這一步，基於服務端啟動繫結埠的NioServerSocketChannel，也就是服務端Channel繫結的NioEventLoop已經啟動。
二、新連線接入
首先我們結合下圖看下當有客戶端接入時，建立NioSocketChannel，然後繫結NioEventLoop並啟動的流程
服務端啟動時會在NioServerSocketChannel的任務鏈中新增ServerBootstrapAcceptor物件，這就是用來處理新新連線接入的
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);
}
// 服務端NioServerSocketChannel的pipeline中新增ServerBootstrapAcceptor
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
在新連線接入事件觸發時，執行unsafe.read();
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
// If the channel implementation throws an exception because there is no event loop, we ignore this
// because we are only trying to determine if ch is registered to this event loop and thus has authority
// to close ch.
return;
}
// Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
// and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
// still healthy and should not be closed.
// See https://github.com/netty/netty/issues/5125
if (eventLoop != this || eventLoop == null) {
return;
}
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
// We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
// the NIO JDK channel implementation may throw a NotYetConnectedException.
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
//新連線接入
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
unsafe.read()的具體實現為NioMessageUnsafe中的read()，在read()方法中主要實現了兩個功能：
1、建立客戶端Channel，也就是NioSocketChannel；
2、開始服務端NioServerSocketChannel的任務鏈傳遞，首先執行之前已經加入任務鏈的ServerBootstrapAcceptor中的channelRead
@Override
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 {
//這裡建立客戶端連線,也就是NioSocketChannelChannel
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
//在這裡開始NioServerSocketChannel的任務鏈傳遞，會首先執行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 {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
接下來在ServerBootstrapAcceptor中的channelRead中會獲取到傳入的NioSocketChannel，針對NioSocketChannel主要會執行以下操作：
1、配置childHandler任務鏈；
2、配置childOptions；
3、為NioSocketChannel分配NioEventLoop
@Override
@SuppressWarnings("unchecked")
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
//配置childHandler任務鏈
child.pipeline().addLast(childHandler);
//配置childOptions
setChannelOptions(child, childOptions, logger);
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
//為新連線分配NioEventLoop，並啟動執行
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);
}
}
看以看到EventLoopGroup中register具體實實現：
1、關於next()，我們之前講過是專門用來分配NioEventLoop；
2、register()主要負責了EventLoop的繫結和啟動；
@Override
public ChannelFuture register(ChannelPromise promise) {
return next().register(promise);
}
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
//與NioEventLoop繫結
AbstractChannel.this.eventLoop = eventLoop;
//首先判斷執行緒是否一致,當前執行緒是NioServerSocketChannel的執行緒,與當前建立NioSocketChannel的eventLoop執行緒不一致
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
//在這裡NioEventLoop啟動
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
上面程式碼中的 eventLoop.execute我們已經分析過，經過一系列的流程，最後會執行NioEventLoop的run方法開始輪詢感興趣的IO事件。
以上我們主要從服務啟動與客戶端連線兩個方面分析了NioEventLoop的啟動流程與原始碼，其實也就對應NioServerSocketChannel與NioSocketChannel分別繫結的NioEventLoop，其中有錯誤和不足之處還請指正與海涵。