BIO到NIO源碼的一些事兒之NIO 上
前言
此篇文章會詳細解讀NIO的功能逐步豐滿的路程，為Reactor-Netty 庫的講解鋪平道路。
關於Java編程方法論-Reactor與Webflux的視頻分享，已經完成了Rxjava 與 Reactor，b站地址如下:
Rxjava源碼解讀與分享：
https://www.bilibili.com/video/av34537840
Reactor源碼解讀與分享：
https://www.bilibili.com/video/av35326911場景代入
接上一篇 BIO到NIO源碼的一些事兒之BIO，我們來接觸NIO的一些事兒。
在上一篇中，我們可以看到，我們要做到非同步非阻塞，我們自己進行的是創建線程池同時對部分代碼做timeout的修改來對接客戶端，但是弊端也很清晰，我們轉換下思維，這裡舉個場景例子，A班同學要和B班同學一起一對一完成任務，每對人拿到的任務是不一樣的，消耗的時間有長有短，任務因為有獎勵所以同學們會搶，傳統模式下，A班同學和B班同學不經管理話，即便只是一個心跳檢測的任務都得一起，在這種情況下，客戶端根本不會有數據要發送，只是想告訴伺服器自己還活著，這種情況下，假如B班再來一個同學做對接的話，就很有問題了，B班的每一個同學都可以看成伺服器端的一個線程。所以，我們需要一個管理者，於是Selector就出現了，作為管理者，這裡，我們往往需要管理同學們的狀態，是否在等待任務，是否在接收信息，是否在輸出信息等等，Selector更側重於動作，針對於這些狀態標籤來做事情就可以了，那這些狀態標籤其實也是需要管理的，於是SelectionKey也就應運而生。接著我們需要對這些同學進行包裝增強，使之攜帶這樣的標籤。同樣，對於同學我們應該進一步解放雙手的，比如給其配臺電腦，這樣，同學是不是可以做更多的事情了，那這個電腦在此處就是Buffer的存在了。 於是在NIO中最主要是有三種角色的，Buffer緩衝區，Channel通道，Selector選擇器，我們都涉及到了，接下來，我們對其源碼一步步分析解讀。
Channel解讀
賦予Channel可非同步可中斷的能力
有上可知，同學其實都是代表著一個個的Socket的存在，那麼這裡Channel就是對其進行的增強包裝，也就是Channel的具體實現裏應該有Socket這個欄位纔行，然後具體實現類裡面也是緊緊圍繞著Socket具備的功能來做文章的。那麼，我們首先來看java.nio.channels.Channel介面的設定:
public interface Channel extends Closeable {
/**
* Tells whether or not this channel is open.
*
* @return {@code true} if, and only if, this channel is open
*/
public boolean isOpen();
/**
* Closes this channel.
*
* <p> After a channel is closed, any further attempt to invoke I/O
* operations upon it will cause a {@link ClosedChannelException} to be
* thrown.
*
* <p> If this channel is already closed then invoking this method has no
* effect.
*
* <p> This method may be invoked at any time. If some other thread has
* already invoked it, however, then another invocation will block until
* the first invocation is complete, after which it will return without
* effect. </p>
*
* @throws IOException If an I/O error occurs
*/
public void close() throws IOException;
}
此處就是很直接的設定，判斷Channel是否是open狀態，關閉Channel的動作，我們在接下來會講到ClosedChannelException是如何具體在代碼中發生的。 有時候，一個Channel可能會被非同步關閉和中斷，這也是我們所需求的。那麼要實現這個效果我們須得設定一個可以進行此操作效果的介面。達到的具體的效果應該是如果線程在實現這個介面的的Channel中進行IO操作的時候，另一個線程可以調用該Channel的close方法。導致的結果就是，進行IO操作的那個阻塞線程會收到一個AsynchronousCloseException異常。
同樣，我們應該考慮到另一種情況，如果線程在實現這個介面的的Channel中進行IO操作的時候，另一個線程可能會調用被阻塞線程的interrupt方法(Thread#interrupt())，從而導致Channel關閉，那麼這個阻塞的線程應該要收到ClosedByInterruptException異常，同時將中斷狀態設定到該阻塞線程之上。
這時候，如果中斷狀態已經在該線程設定完畢，此時在其之上的有Channel又調用了IO阻塞操作，那麼，這個Channel會被關閉，同時，該線程會立即受到一個ClosedByInterruptException異常，它的interrupt狀態仍然保持不變。 這個介面定義如下:
public interface InterruptibleChannel
extends Channel
{
/**
* Closes this channel.
*
* <p> Any thread currently blocked in an I/O operation upon this channel
* will receive an {@link AsynchronousCloseException}.
*
* <p> This method otherwise behaves exactly as specified by the {@link
* Channel#close Channel} interface. </p>
*
* @throws IOException If an I/O error occurs
*/
public void close() throws IOException;
}
其針對上面所提到邏輯的具體實現是在java.nio.channels.spi.AbstractInterruptibleChannel進行的，關於這個類的解析，我們來參考這篇文章InterruptibleChannel 與可中斷 IO
賦予Channel可被多路復用的能力
我們在前面有說到，Channel可以被Selector進行使用，而Selector是根據Channel的狀態來分配任務的，那麼Channel應該提供一個註冊到Selector上的方法，來和Selector進行綁定。也就是說Channel的實例要調用register(Selector,int,Object)。注意，因為Selector是要根據狀態值進行管理的，所以此方法會返回一個SelectionKey對象來表示這個channel在selector上的狀態。關於SelectionKey，它是包含很多東西的，這裡暫不提。
//java.nio.channels.spi.AbstractSelectableChannel#register
public final SelectionKey register(Selector sel, int ops, Object att)
throws ClosedChannelException
{
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (!isOpen())
throw new ClosedChannelException();
synchronized (regLock) {
if (isBlocking())
throw new IllegalBlockingModeException();
synchronized (keyLock) {
// re-check if channel has been closed
if (!isOpen())
throw new ClosedChannelException();
SelectionKey k = findKey(sel);
if (k != null) {
k.attach(att);
k.interestOps(ops);
} else {
// New registration
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
return k;
}
}
}
//java.nio.channels.spi.AbstractSelectableChannel#addKey
private void addKey(SelectionKey k) {
assert Thread.holdsLock(keyLock);
int i = 0;
if ((keys != null) && (keyCount < keys.length)) {
// Find empty element of key array
for (i = 0; i < keys.length; i++)
if (keys[i] == null)
break;
} else if (keys == null) {
keys = new SelectionKey[2];
} else {
// Grow key array
int n = keys.length * 2;
SelectionKey[] ks = new SelectionKey[n];
for (i = 0; i < keys.length; i++)
ks[i] = keys[i];
keys = ks;
i = keyCount;
}
keys[i] = k;
keyCount++;
}
一旦註冊到Selector上，Channel將一直保持註冊直到其被解除註冊。在解除註冊的時候會解除Selector分配給Channel的所有資源。 也就是Channel並沒有直接提供解除註冊的方法，那我們換一個思路，我們將Selector上代表其註冊的Key取消不就可以了。這裡可以通過調用SelectionKey#cancel()方法來顯式的取消key。然後在Selector下一次選擇操作期間進行對Channel的取消註冊。
//java.nio.channels.spi.AbstractSelectionKey#cancel
/**
* Cancels this key.
*
* <p> If this key has not yet been cancelled then it is added to its
* selectors cancelled-key set while synchronized on that set. </p>
*/
public final void cancel() {
// Synchronizing "this" to prevent this key from getting canceled
// multiple times by different threads, which might cause race
// condition between selectors select() and channels close().
synchronized (this) {
if (valid) {
valid = false;
//還是調用Selector的cancel方法
((AbstractSelector)selector()).cancel(this);
}
}
}
//java.nio.channels.spi.AbstractSelector#cancel
void cancel(SelectionKey k) {
synchronized (cancelledKeys) {
cancelledKeys.add(k);
}
}
//在下一次select操作的時候來解除那些要求cancel的key，即解除Channel註冊
//sun.nio.ch.SelectorImpl#select(long)
@Override
public final int select(long timeout) throws IOException {
if (timeout < 0)
throw new IllegalArgumentException("Negative timeout");
//重點關注此方法
return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
}
//sun.nio.ch.SelectorImpl#lockAndDoSelect
private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout)
throws IOException
{
synchronized (this) {
ensureOpen();
if (inSelect)
throw new IllegalStateException("select in progress");
inSelect = true;
try {
synchronized (publicSelectedKeys) {
//重點關注此方法
return doSelect(action, timeout);
}
} finally {
inSelect = false;
}
}
}
//sun.nio.ch.WindowsSelectorImpl#doSelect
protected int doSelect(Consumer<SelectionKey> action, long timeout)
throws IOException
{
assert Thread.holdsLock(this);
this.timeout = timeout; // set selector timeout
processUpdateQueue();
//重點關注此方法
processDeregisterQueue();
if (interruptTriggered) {
resetWakeupSocket();
return 0;
}
...
}
/**
* sun.nio.ch.SelectorImpl#processDeregisterQueue
* Invoked by selection operations to process the cancelled-key set
*/
protected final void processDeregisterQueue() throws IOException {
assert Thread.holdsLock(this);
assert Thread.holdsLock(publicSelectedKeys);
Set<SelectionKey> cks = cancelledKeys();
synchronized (cks) {
if (!cks.isEmpty()) {
Iterator<SelectionKey> i = cks.iterator();
while (i.hasNext()) {
SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
i.remove();
// remove the key from the selector
implDereg(ski);
selectedKeys.remove(ski);
keys.remove(ski);
// remove from channels key set
deregister(ski);
SelectableChannel ch = ski.channel();
if (!ch.isOpen() && !ch.isRegistered())
((SelChImpl)ch).kill();
}
}
}
}
這裡，當Channel關閉時，無論是通過調用Channel#close還是通過打斷線程的方式來對Channel進行關閉，其都會隱式的取消關於這個Channel的所有的keys，其內部也是調用了k.cancel()。
//java.nio.channels.spi.AbstractInterruptibleChannel#close
/**
* Closes this channel.
*
* <p> If the channel has already been closed then this method returns
* immediately. Otherwise it marks the channel as closed and then invokes
* the {@link #implCloseChannel implCloseChannel} method in order to
* complete the close operation. </p>
*
* @throws IOException
* If an I/O error occurs
*/
public final void close() throws IOException {
synchronized (closeLock) {
if (closed)
return;
closed = true;
implCloseChannel();
}
}
//java.nio.channels.spi.AbstractSelectableChannel#implCloseChannel
protected final void implCloseChannel() throws IOException {
implCloseSelectableChannel();
// clone keys to avoid calling cancel when holding keyLock
SelectionKey[] copyOfKeys = null;
synchronized (keyLock) {
if (keys != null) {
copyOfKeys = keys.clone();
}
}
if (copyOfKeys != null) {
for (SelectionKey k : copyOfKeys) {
if (k != null) {
k.cancel(); // invalidate and adds key to cancelledKey set
}
}
}
}
如果Selector自身關閉掉，那麼Channel也會被解除註冊，同時代表Channel註冊的key也將變得無效：
//java.nio.channels.spi.AbstractSelector#close
public final void close() throws IOException {
boolean open = selectorOpen.getAndSet(false);
if (!open)
return;
implCloseSelector();
}
//sun.nio.ch.SelectorImpl#implCloseSelector
@Override
public final void implCloseSelector() throws IOException {
wakeup();
synchronized (this) {
implClose();
synchronized (publicSelectedKeys) {
// Deregister channels
Iterator<SelectionKey> i = keys.iterator();
while (i.hasNext()) {
SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
deregister(ski);
SelectableChannel selch = ski.channel();
if (!selch.isOpen() && !selch.isRegistered())
((SelChImpl)selch).kill();
selectedKeys.remove(ski);
i.remove();
}
assert selectedKeys.isEmpty() && keys.isEmpty();
}
}
}
一個channel最多可以最多隻能在特定的selector註冊一次。我們可以通過調用java.nio.channels.SelectableChannel#isRegistered的方法來確定是否向一個或多個Selector註冊了channel。
//java.nio.channels.spi.AbstractSelectableChannel#isRegistered
// -- Registration --
public final boolean isRegistered() {
synchronized (keyLock) {
//我們在之前往Selector上註冊的時候調用了addKey方法，即每次往//一個Selector註冊一次，keyCount就要自增一次。
return keyCount != 0;
}
}
至此，繼承了SelectableChannel這個類之後，這個channel就可以安全的由多個並發線程來使用。 這裡，要注意的是，繼承了AbstractSelectableChannel這個類之後，新創建的channel始終處於阻塞模式。然而與Selector的多路復用有關的操作必須基於非阻塞模式，所以在註冊到Selector之前，必須將channel置於非阻塞模式，並且在取消註冊之前，channel可能不會返回到阻塞模式。 這裡，我們涉及了Channel的阻塞模式與非阻塞模式。在阻塞模式下，在Channel上調用的每個I/O操作都將阻塞，直到完成為止。 在非阻塞模式下，I/O操作永遠不會阻塞，並且可以傳輸比請求的位元組更少的位元組，或者根本不傳輸任何位元組。 我們可以通過調用channel的isBlocking方法來確定其是否為阻塞模式。
//java.nio.channels.spi.AbstractSelectableChannel#register
public final SelectionKey register(Selector sel, int ops, Object att)
throws ClosedChannelException
{
if ((ops & ~validOps()) != 0)
throw new IllegalArgumentException();
if (!isOpen())
throw new ClosedChannelException();
synchronized (regLock) {
//此處會做判斷，假如是阻塞模式，則會返回true，然後就會拋出異常
if (isBlocking())
throw new IllegalBlockingModeException();
synchronized (keyLock) {
// re-check if channel has been closed
if (!isOpen())
throw new ClosedChannelException();
SelectionKey k = findKey(sel);
if (k != null) {
k.attach(att);
k.interestOps(ops);
} else {
// New registration
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
return k;
}
}
}
所以，我們在使用的時候可以基於以下的例子作為參考:
public NIOServerSelectorThread(int port)
{
try {
//打開ServerSocketChannel，用於監聽客戶端的連接，他是所有客戶端連接的父管道
serverSocketChannel = ServerSocketChannel.open();
//將管道設置為非阻塞模式
serverSocketChannel.configureBlocking(false);
//利用ServerSocketChannel創建一個服務端Socket對象，即ServerSocket
serverSocket = serverSocketChannel.socket();
//為服務端Socket綁定監聽埠
serverSocket.bind(new InetSocketAddress(port));
//創建多路復用器
selector = Selector.open();
//將ServerSocketChannel註冊到Selector多路復用器上，並且監聽ACCEPT事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("The server is start in port: "+port);
} catch (IOException e) {
e.printStackTrace();
}
}
因時間關係，本篇暫時到這裡，剩下的會在下一篇中進行講解。
推薦閱讀：