Netty 源码解析 -- 事件循环机制实现原理

while(true) {
	process(selector.select());

	process(getTask());
}

if (eventLoop.inEventLoop()) {
	register0(promise);	
} else {
	eventLoop.execute(new Runnable() {	
		public void run() {
			register0(promise);
		}
	});
}

private final Queue<Runnable> taskQueue;	// 待处理异步任务
private volatile Thread thread;				// EventLoop执行线程，即SingleThreadEventExecutor创建的新线程
private final Executor executor;			// java.util.concurrent.Executor，负责创建线程

public void execute(Runnable task) {
	...

	boolean inEventLoop = inEventLoop();
	// #1
	addTask(task);
	// #2
	if (!inEventLoop) {
		startThread();
		// #3
		if (isShutdown()) {
			...
		}
	}
	// #4
	if (!addTaskWakesUp && wakesUpForTask(task)) {
		wakeup(inEventLoop);
	}
}

protected void run() {
	int selectCnt = 0;
	// #1
	for (;;) {

			int strategy;
			
				// #2
				strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
				switch (strategy) {
				case SelectStrategy.CONTINUE:
					continue;

				case SelectStrategy.BUSY_WAIT:
					// fall-through to SELECT since the busy-wait is not supported with NIO

				case SelectStrategy.SELECT:
					// #3
					long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
					if (curDeadlineNanos == -1L) {
						curDeadlineNanos = NONE; // nothing on the calendar
					}
					nextWakeupNanos.set(curDeadlineNanos);
					try {
						// #4
						if (!hasTasks()) {
							strategy = select(curDeadlineNanos);
						}
					} finally {
						// #5
						nextWakeupNanos.lazySet(AWAKE);
					}
					// fall through
				default:
				}
				...
			
			// #6
			selectCnt++;
			cancelledKeys = 0;
			needsToSelectAgain = false;
			final int ioRatio = this.ioRatio;
			boolean ranTasks;
			// #7
			if (ioRatio == 100) {
				try {
					if (strategy > 0) {
						processSelectedKeys();
					}
				} finally {
					// Ensure we always run tasks.
					ranTasks = runAllTasks();
				}
			} else if (strategy > 0) {
				final long ioStartTime = System.nanoTime();
				try {
					processSelectedKeys();
				} finally {
					// Ensure we always run tasks.
					final long ioTime = System.nanoTime() - ioStartTime;
					ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
				}
			} else {
				ranTasks = runAllTasks(0); // This will run the minimum number of tasks
			}
			// #8
			if (ranTasks || strategy > 0) {
				if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
					logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
							selectCnt - 1, selector);
				}
				selectCnt = 0;
			} else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
				selectCnt = 0;
			}
		
		
			// #9
			if (isShuttingDown()) {
				closeAll();
				if (confirmShutdown()) {
					return;
				}
			}
		
	}
}

Selector unwrappedSelector;				// JVM中的Selector
Selector selector;						// 优化后的SelectedSelectionKeySetSelector
SelectedSelectionKeySet selectedKeys;	// 对(jvm)Selector#selectedKeys进行优化

private SelectorTuple openSelector() {
    final Selector unwrappedSelector;
    try {
		// #1
        unwrappedSelector = provider.openSelector();
    } catch (IOException e) {
        throw new ChannelException("failed to open a new selector", e);
    }

    if (DISABLE_KEY_SET_OPTIMIZATION) {
        return new SelectorTuple(unwrappedSelector);
    }

    ...

    final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
    final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();

    Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
        
        public Object run() {
            try {
				// #2
                Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
                Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");

                ...

                selectedKeysField.set(unwrappedSelector, selectedKeySet);
                publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
                return null;
            } ...
        }
    });

    ...
    selectedKeys = selectedKeySet;
    logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector);
	// #3
    return new SelectorTuple(unwrappedSelector,
                             new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));
}

private int select(long deadlineNanos) throws IOException {
	// #1
	if (deadlineNanos == NONE) {
		return selector.select();
	}

	// #2
	long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
	return timeoutMillis <= 0 ? selector.selectNow() : selector.select(timeoutMillis);
}

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
	final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
	...

	try {
		int readyOps = k.readyOps();
		// #1
		if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
			int ops = k.interestOps();
			ops &= ~SelectionKey.OP_CONNECT;
			k.interestOps(ops);

			unsafe.finishConnect();
		}

		// #2
		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
		// #3
		if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
			unsafe.read();
		}
	} catch (CancelledKeyException ignored) {
		unsafe.close(unsafe.voidPromise());
	}
}

protected boolean runAllTasks(long timeoutNanos) {
    // #1
    fetchFromScheduledTaskQueue();
    Runnable task = pollTask();
    if (task == null) {
        afterRunningAllTasks();
        return false;
    }

    final long deadline = timeoutNanos > 0 ? ScheduledFutureTask.nanoTime() + timeoutNanos : 0;
    long runTasks = 0;
    long lastExecutionTime;
    for (;;) {
    	// #2
        safeExecute(task);

        runTasks ++;

        // #3
        if ((runTasks & 0x3F) == 0) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            if (lastExecutionTime >= deadline) {
                break;
            }
        }
        // #4
        task = pollTask();
        if (task == null) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            break;
        }
    }
    // #5
    afterRunningAllTasks();
    this.lastExecutionTime = lastExecutionTime;
    return true;
}