對NIO的理解
個人單方面認為，NIO與BIO的最大區別在於主動和被動，使用BIO的方式需要等待被調用方返回數據，很明顯此時調用者是被動的。
舉個例子
阻塞IO
假設你是一個膽小又害羞的男孩子，你約了隔壁測試的妹子，但你並不敢主動約會，所以你把自己的手機號碼給她，並暗示她想要約會的時候打電話給你。很明顯此時你陷入了被動，約不約會的結果需要妹子主動告知你，如果她忘了，那么你要陷入長時間的等待中以及無盡的猜測和自我懷疑中（太慘了)。[如果你是一個膽小害羞又好色的男孩子，那就慘了]
非阻塞IO 我們知道，渣男通常有很多的備胎，我管這個叫做備胎池(SpareTirePool), 那么當他想要約會的時候，只要群發問妹子要不要約會，如果要約會的話就和妹子約會，約會結束之后，處理其他約會事件，如果沒有繼續下一次詢問。在這個例子中約會可以視為IO事件，問妹子的過程可以視為備胎池的輪詢。
如果你要學習NIO，可以學習
Tomcat 如何使用NIO
既然是網絡通信的I/O那必然有以下兩個步驟
- SeverSocket的啟動
- I/O事件的處理
關鍵代碼在 package org.apache.tomcat.util.net.NioEndpoint 中
P.S. 文章太長，如果不想看可以直接閱讀結論
ServerSocket的啟動
在最開始看代碼，是震驚的，真的，如果你看Reactor模型的話
以下bind方法代碼是啟動ServerSocket的流程，主要流程如下
- 綁定地址
- 設置接收新連接的方式為阻塞方式(關鍵點)
- 設置Acceptor和Poller的數量以及初始化SelectorPool
@Override
public void bind() throws Exception {
if (!getUseInheritedChannel()) {
serverSock = ServerSocketChannel.open();
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
serverSock.socket().bind(addr,getAcceptCount());
} else {
// Retrieve the channel provided by the OS
Channel ic = System.inheritedChannel();
if (ic instanceof ServerSocketChannel) {
serverSock = (ServerSocketChannel) ic;
}
if (serverSock == null) {
throw new IllegalArgumentException(sm.getString("endpoint.init.bind.inherited"));
}
}
// 以阻塞的方式來接收連接!!
serverSock.configureBlocking(true); //mimic APR behavior
// 設置Acceptor和Poller的數量
if (acceptorThreadCount == 0) {
// FIXME: Doesn't seem to work that well with multiple accept threads
// 顧名思義,Acceptor是用來處理新連接的
acceptorThreadCount = 1;
}
if (pollerThreadCount <= 0) {
// Poller 用來處理I/O事件
pollerThreadCount = 1;
}
setStopLatch(new CountDownLatch(pollerThreadCount));
// Initialize SSL if needed
initialiseSsl();
// 從此處可以看出tomcat池化了selector
selectorPool.open();
}
Tomcat NIO 如何處理I/O事件
先說結論，Tomcat NIO模型中有以下關鍵角色
- Acceptor 用於接收新連接，每個Acceptor一個線程，以阻塞的方式接收新連接
- Poller 當Acceptor接收到新連接，進行處理之后選擇一個Poller處理該連接上的I/O事件。
- LimitLatch 一個用來限制連接數的鎖
Acceptor
Acceptor的主要工作就是不斷接收來自客戶端的連接，在簡單處理之后將該連接交給Poller處理
接收來自客戶端連接, 如果你不想看代碼，以下是其主要流程
- 接收來自客戶端的連接，並將其交給Poller處理
@Override
public void run() {
int errorDelay = 0;
// running的檢測貫穿了Accpetor的處理流程，在每次關鍵操作的時候都會執行檢測
while (running) {
// 如果進入暫停狀態則每隔一段時間檢測一下
while (paused && running) {
state = AcceptorState.PAUSED;
try {
Thread.sleep(50);
} catch (InterruptedException e) {
// Ignore
}
}
// 再次檢測
if (!running) {
break;
}
state = AcceptorState.RUNNING;
try {
//檢查是否達到最大連接數如果是則陷入等待，如果不是則增加當前連接數
countUpOrAwaitConnection();
SocketChannel socket = null;
try {
//接收新連接
socket = serverSock.accept();
} catch (IOException ioe) {
// 發生異常,則減少連接數
countDownConnection();
if (running) {
handleExceptionWithDelay(errorDelay);
// re-throw
throw ioe;
} else {
break;
}
}
// Successful accept, reset the error delay
errorDelay = 0;
// Configure the socket
if (running && !paused) {
//setSocketOptions會導致將該連接交給Poller處理
if (!setSocketOptions(socket)) {
closeSocket(socket);
}
} else {
closeSocket(socket);
}
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error(sm.getString("endpoint.accept.fail"), t);
}
}
state = AcceptorState.ENDED;
}
再來看看setSocketOptions做了什么,不想看代碼的話，總結如下
- 將客戶端socket設置為非阻塞模式
- 將客戶端的socket封裝為NioChannel或SecureNioChannel(使用了對象池技術)
- 從Poller池中獲取一個Poller，將NioChannel注冊到Poller上
protected boolean setSocketOptions(SocketChannel socket) {
// Process the connection
try {
//設置為非阻塞模式，以便通過selector進行查詢
socket.configureBlocking(false);
Socket sock = socket.socket();
socketProperties.setProperties(sock);
//從對象池中獲取一個NioChannel，tomcat會復用一切可以復用的對象以減少創建新對象所帶來的消耗
NioChannel channel = nioChannels.pop();
if (channel == null) {
// 沒有獲取到，那就新建一個唄
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
// SSL這一塊還沒研究
if (isSSLEnabled()) {
channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
} else {
channel = new NioChannel(socket, bufhandler);
}
} else {
channel.setIOChannel(socket);
//重新設置SocketBufferHandler，將其設置為可寫和可讀
channel.reset();
}
//從Poller池中獲取一個Poller(按照次序獲取,可以理解為一個圓環)，並將Channel注冊到上面
getPoller0().register(channel);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
try {
log.error("",t);
} catch (Throwable tt) {
ExceptionUtils.handleThrowable(tt);
}
// Tell to close the socket
return false;
}
return true;
}
Poller
從連接注冊到Poller說起
不加鎖的獲取一個Poller
具體說明見代碼
關鍵點：對一個數A取余會將余數的結果限制在A的范圍內
/**
* Return an available poller in true round robin fashion.
* 很明顯，取余的方式揭示了獲取Poller的方法。你可以理解為
* Poller會組成一個圓環，這樣我們就可以通過不斷遞增獲取
* 下一個Poller,但是數據會溢出所以我們要取絕對值
* @return The next poller in sequence
*/
public Poller getPoller0() {
int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
return pollers[idx];
}
channel的注冊
該方法會對新的建的連接進行封裝，並以PollerEvent的形式注冊到相應的Poller中
需要注意的是，真正的注冊讀事件並不是在此方法注冊的(當前方法調用者為Acceptor線程)，而是在Poller線程中注冊讀事件的
/**
* Registers a newly created socket with the poller.
* 將新建的socket注冊到Poller上
* @param socket The newly created socket
*/
public void register(final NioChannel socket) {
//以下代碼為設置各種參數，可以從方法名進行推測，不再進行敘述
socket.setPoller(this);
NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
socket.setSocketWrapper(ka);
ka.setPoller(this);
ka.setReadTimeout(getSocketProperties().getSoTimeout());
ka.setWriteTimeout(getSocketProperties().getSoTimeout());
ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
ka.setSecure(isSSLEnabled());
ka.setReadTimeout(getConnectionTimeout());
ka.setWriteTimeout(getConnectionTimeout());
//從緩存中獲取一個PollerEvent
PollerEvent r = eventCache.pop();
// 注冊讀事件
ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
// 如果沒有從緩存中獲取，那么就新建一個
if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
else r.reset(socket,ka,OP_REGISTER);
addEvent(r);
}
Poller處理I/O 事件
Poller 處理I/O事件的的代碼較長，而且細節也較多，總結其主要作用如下
- 檢測是否有Acceptor提交PollerEvent，如果有則調用PolllerEvent的run方法注冊讀事件
- 在執行關鍵操作的時候檢測該Poller是否被關閉如果是，則執行相應的資源釋放和關閉操作
- 調用selector.select() 輪詢事件，如果有讀事件則交給processKey處理
@Override
public void run() {
// Loop until destroy() is called
// 一直循環直到destroy方法被調用
while (true) {
boolean hasEvents = false;
try {
if (!close) {
// events 方法會處理Acceptor注冊到Poller中的PollerEvent
// 主要是注冊讀事件
hasEvents = events();
if (wakeupCounter.getAndSet(-1) > 0) {
//if we are here, means we have other stuff to do
//do a non blocking select
keyCount = selector.selectNow();
} else {
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
// 檢測到關閉，則處理剩余的事件並關閉selector
if (close) {
// 處理Acceptors注冊到Poller中的PollerEvent
events();
//selector time out 或者poller被關閉就會調用timeout方法
timeout(0, false);
try {
selector.close();
} catch (IOException ioe) {
log.error(sm.getString("endpoint.nio.selectorCloseFail"), ioe);
}
break;
}
} catch (Throwable x) {
ExceptionUtils.handleThrowable(x);
log.error("",x);
continue;
}
//either we timed out or we woke up, process events first
if ( keyCount == 0 ) hasEvents = (hasEvents | events());
// 執行 select 操作，查詢I/O事件
Iterator<SelectionKey> iterator =
keyCount > 0 ? selector.selectedKeys().iterator() : null;
// Walk through the collection of ready keys and dispatch
// any active event.
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (attachment == null) {
iterator.remove();
} else {
iterator.remove();
// 處理檢測到的I/O事件
processKey(sk, attachment);
}
}//while
//timeout 會檢查是否關閉，如果已經關閉並且有事件未處理會調用cancelledKey方法
//cancelledKey:該方法主要是對和該連接相關的資源執行關閉操作
timeout(keyCount,hasEvents);
}//while
getStopLatch().countDown();
}
processKey 處理I/O事件
processKey主要工作如下
- 再次檢測Poller是否關閉，如果是則釋放資源
- 檢測查詢到事件是否合法，如果合法則取消已注冊到selector上的事件且被被本次輪詢所查詢到的事件
- 再調用processSocket處理讀事件，之后處理寫事件
protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
try {
if ( close ) {
// 如果Poller關閉則關閉和釋放和此連接相關的資源
cancelledKey(sk);
} else if ( sk.isValid() && attachment != null ) {
if (sk.isReadable() || sk.isWritable() ) {
if ( attachment.getSendfileData() != null ) {
processSendfile(sk,attachment, false);
} else {
// 取消注冊事件
// sk.interestOps()& (~readyOps)
unreg(sk, attachment, sk.readyOps());
boolean closeSocket = false;
// Read goes before write 先讀后寫
if (sk.isReadable()) {
// 關鍵代碼，調用processSocket方法處理讀事件
if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
if (!closeSocket && sk.isWritable()) {
if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
if (closeSocket) {
cancelledKey(sk);
}
}
}
} else {
//invalid key
cancelledKey(sk);
}
} catch ( CancelledKeyException ckx ) {
cancelledKey(sk);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error("",t);
}
}
processSocket 真-處理I/O事件
processSocket定義在org.apache.tomcat.util.net.AbstractEndPoint中, 也就是意味着無論你采用的是BIO還是NIO或者NIO2最終讀寫數據都是調用此方法
從代碼中可以看出，依然是對象池，依然是再次封裝（套娃），並將其提交到線程池中執行，接下來的內容就不再本次討論范圍內呢。
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
return false;
}
SocketProcessorBase<S> sc = processorCache.pop();
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
} catch (RejectedExecutionException ree) {
getLog().warn(sm.getString("endpoint.executor.fail", socketWrapper) , ree);
return false;
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
// This means we got an OOM or similar creating a thread, or that
// the pool and its queue are full
getLog().error(sm.getString("endpoint.process.fail"), t);
return false;
}
return true;
}
總結
Tomcat的NIO模型
手抖了，線不怎么♂
LimitLatch 為所有的Acceptor共用，用來限制當前的最大連接數
Acceptor 以阻塞的形式來接收新連接，並將其封裝成PollerEvent對象提交到Poller中
Poller 接收來自Acceptor的PollerEvent並注冊讀事件，以及輪詢和其綁定的客戶端Socket有無讀事件，如果有則執行進一步操作，將其提交到其他地方執行處理(解析Http協議)
思想遷移
學習源碼就是為了學習其設計思想. -- 沃茲及.碩德
對象池化 池化對象、池化連接可以大大降低新建對象以及GC所帶來的消耗，當需要使用從池中取出來重新設置相關值即可
環形隊列 雖然這玩意不新鮮，但配合上原子類，就可以在高並發的情況，高效的獲取隊列中的下一個元素（環形隊列中索引溢出的處理在之前我是沒有考慮到的)
阻塞獲取鏈接，非阻塞處理IO事件 與Reactor模型形成強烈的對比，學習NIO的時候思維被限制住了，認為非阻塞的獲取連接會獲得更高的性能，但現在情況不一定了（還沒測試，哪位老哥試了告訴我一下)
關鍵操作時，對標志位進行檢測 如果你要通過一個標志變量來控制你的線程，且線程循環一次需要相對較長的時間（你代碼太長，操作太多）那么最好在執行關鍵操作之前對你的標志變量進行檢查，來決定是否要改變線程的行為（康康poller和Acceptor的代碼)
初次學習Tomcat的代碼，有理解錯誤的地方還請大佬指出