BIO到NIO原始碼的一些事兒之NIO 中
阿新 • • 發佈：2019-01-09
前言
此係列文章會詳細解讀NIO的功能逐步豐滿的路程，為Reactor-Netty 庫的講解鋪平道路。
關於Java程式設計方法論-Reactor與Webflux的視訊分享，已經完成了Rxjava 與 Reactor，b站地址如下:
Rxjava原始碼解讀與分享：www.bilibili.com/video/av345…
Reactor原始碼解讀與分享：www.bilibili.com/video/av353…
本系列原始碼解讀基於JDK11 api細節可能與其他版本有所差別，請自行解決jdk版本問題。
Channel解讀
賦予Channel支援網路socket的能力
我們最初的目的就是為了增強Socket，基於這個基本需求，沒有條件創造條件，於是為了讓Channel擁有網路socket的能力，這裡定義了一個java.nio.channels.NetworkChannel介面。花不多說，我們來看這個介面的定義:
public interface NetworkChannel extends Channel
{
NetworkChannel bind(SocketAddress local) throws IOException;
SocketAddress getLocalAddress() throws IOException;
<T> NetworkChannel setOption 通過bind(SocketAddress) 方法將socket繫結到本地 SocketAddress上，通過getLocalAddress()方法返回socket繫結的地址， 通過 setOption(SocketOption,Object)
getOption(SocketOption)方法設定和查詢socket支援的配置選項。
bind
接下來我們來看 java.nio.channels.ServerSocketChannel抽象類及其實現類sun.nio.ch.ServerSocketChannelImpl對之實現的細節。 首先我們來看其對於bind的實現:
//sun.nio.ch.ServerSocketChannelImpl#bind
@Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
synchronized (stateLock) {
ensureOpen();
//通過localAddress判斷是否已經呼叫過bind
if (localAddress != null)
throw new AlreadyBoundException();
//InetSocketAddress(0)表示繫結到本機的所有地址，由作業系統選擇合適的埠
InetSocketAddress isa = (local == null)
? new InetSocketAddress(0)
: Net.checkAddress(local);
SecurityManager sm = System.getSecurityManager();
if (sm != null)
sm.checkListen(isa.getPort());
NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
Net.bind(fd, isa.getAddress(), isa.getPort());
//開啟監聽，s如果引數backlog小於1，預設接受50個連線
Net.listen(fd, backlog < 1 ? 50 : backlog);
localAddress = Net.localAddress(fd);
}
return this;
}
複製程式碼下面我們來看看Net中的bind和listen方法是如何實現的。
Net.bind
//sun.nio.ch.Net#bind(java.io.FileDescriptor, java.net.InetAddress, int)
public static void bind(FileDescriptor fd, InetAddress addr, int port)
throws IOException
{
bind(UNSPEC, fd, addr, port);
}
static void bind(ProtocolFamily family, FileDescriptor fd,
InetAddress addr, int port) throws IOException
{
//如果傳入的協議域不是IPV4而且支援IPV6,則使用ipv6
boolean preferIPv6 = isIPv6Available() &&
(family != StandardProtocolFamily.INET);
bind0(fd, preferIPv6, exclusiveBind, addr, port);
}
private static native void bind0(FileDescriptor fd, boolean preferIPv6,
boolean useExclBind, InetAddress addr,
int port)
throws IOException;
複製程式碼bind0為native方法實現:
JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_bind0(JNIEnv *env, jclass clazz, jobject fdo, jboolean preferIPv6,
jboolean useExclBind, jobject iao, int port)
{
SOCKETADDRESS sa;
int sa_len = 0;
int rv = 0;
//將java的InetAddress轉換為c的struct sockaddr
if (NET_InetAddressToSockaddr(env, iao, port, &sa, &sa_len,
preferIPv6) != 0) {
return;//轉換失敗，方法返回
}
//呼叫bind方法:int bind(int sockfd, struct sockaddr* addr, socklen_t addrlen)
rv = NET_Bind(fdval(env, fdo), &sa, sa_len);
if (rv != 0) {
handleSocketError(env, errno);
}
}
複製程式碼socket是使用者程式與核心互動資訊的樞紐，它自身沒有網路協議地址和埠號等資訊,在進行網路通訊的時候，必須把一個socket與一個地址相關聯。 很多時候核心會我們自動繫結一個地址，然而有時使用者可能需要自己來完成這個繫結的過程，以滿足實際應用的需要； 最典型的情況是一個伺服器程序需要繫結一個眾所周知的地址或埠以等待客戶來連線。 對於客戶端，很多時候並不需要呼叫bind方法，而是由核心自動繫結；
這裡要注意，繫結歸繫結，在有連線過來的時候會建立一個新的Socket，然後服務端操作這個新的Socket即可。這裡就可以關注accept方法了。由sun.nio.ch.ServerSocketChannelImpl#bind最後，我們知道其通過Net.listen(fd, backlog < 1 ? 50 : backlog)開啟監聽，如果引數backlog小於1，預設接受50個連線。由此，我們來關注下Net.listen方法細節。
Net.listen
//sun.nio.ch.Net#listen
static native void listen(FileDescriptor fd, int backlog) throws IOException;
複製程式碼可以知道，Net.listen是native方法，原始碼如下：
JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_listen(JNIEnv *env, jclass cl, jobject fdo, jint backlog)
{
if (listen(fdval(env, fdo), backlog) < 0)
handleSocketError(env, errno);
}
複製程式碼可以看到底層是呼叫listen實現的，listen函式在一般在呼叫bind之後到呼叫accept之前呼叫，它的函式原型是： int listen(int sockfd, int backlog)返回值：0表示成功， -1表示失敗
我們再來關注下bind操作中的其他細節，最開始時的ensureOpen()方法判斷:
//sun.nio.ch.ServerSocketChannelImpl#ensureOpen
// @throws ClosedChannelException if channel is closed
private void ensureOpen() throws ClosedChannelException {
if (!isOpen())
throw new ClosedChannelException();
}
//java.nio.channels.spi.AbstractInterruptibleChannel#isOpen
public final boolean isOpen() {
return !closed;
}
複製程式碼如果socket關閉，則丟擲ClosedChannelException 。
我們再來看下Net#checkAddress：
//sun.nio.ch.Net#checkAddress(java.net.SocketAddress)
public static InetSocketAddress checkAddress(SocketAddress sa) {
if (sa == null)//地址為空
throw new NullPointerException();
//非InetSocketAddress型別地址
if (!(sa instanceof InetSocketAddress))
throw new UnsupportedAddressTypeException(); // ## needs arg
InetSocketAddress isa = (InetSocketAddress)sa;
//地址不可識別
if (isa.isUnresolved())
throw new UnresolvedAddressException(); // ## needs arg
InetAddress addr = isa.getAddress();
//非ip4和ip6地址
if (!(addr instanceof Inet4Address || addr instanceof Inet6Address))
throw new IllegalArgumentException("Invalid address type");
return isa;
}
複製程式碼從上面可以看出，bind首先檢查ServerSocket是否關閉，是否繫結地址， 如果既沒有繫結也沒關閉，則檢查繫結的socketaddress是否正確或合法； 然後通過Net工具類的bind和listen，完成實際的ServerSocket地址繫結和開啟監聽，如果繫結是開啟的引數小於1，則預設接受50個連線。
對照我們之前在第一篇中接觸的BIO，我們來看些accept()方法的實現:
//sun.nio.ch.ServerSocketChannelImpl#accept()
@Override
public SocketChannel accept() throws IOException {
acceptLock.lock();
try {
int n = 0;
FileDescriptor newfd = new FileDescriptor();
InetSocketAddress[] isaa = new InetSocketAddress[1];
boolean blocking = isBlocking();
try {
begin(blocking);
do {
n = accept(this.fd, newfd, isaa);
} while (n == IOStatus.INTERRUPTED && isOpen());
} finally {
end(blocking, n > 0);
assert IOStatus.check(n);
}
if (n < 1)
return null;
//針對接受連線的處理通道socketchannelimpl，預設為阻塞模式
// newly accepted socket is initially in blocking mode
IOUtil.configureBlocking(newfd, true);
InetSocketAddress isa = isaa[0];
//構建SocketChannelImpl，這個具體在SocketChannelImpl再說
SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
// check permitted to accept connections from the remote address
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
//檢查地址和port許可權
sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
} catch (SecurityException x) {
sc.close();
throw x;
}
}
//返回socketchannelimpl
return sc;
} finally {
acceptLock.unlock();
}
}
複製程式碼對於accept(this.fd, newfd, isaa)，呼叫accept接收socket中已建立的連線，我們之前有在BIO中瞭解過，函式最終會呼叫:int accept(int sockfd,struct sockaddr *addr, socklen_t *addrlen);
- 如果fd監聽socket的佇列中沒有等待的連線，socket也沒有被標記為Non-blocking，accept()會阻塞直到連接出現；
- 如果socket被標記為Non-blocking，佇列中也沒有等待的連線，accept()返回錯誤EAGAIN或EWOULDBLOCK
這裡begin(blocking); 與 end(blocking, n > 0);的合作模式我們在InterruptibleChannel 與可中斷 IO這一篇文章中已經涉及過，這裡再次提一下，讓大家看到其應用，此處專注的是等待連線這個過程，期間可以出現異常打斷，這個過程正常結束的話，就會正常往下執行邏輯，不要搞的好像這個Channel要結束了一樣，end(blocking, n > 0)的第二個引數completed也只是在判斷這個等待過程是否結束而已，不要功能範圍擴大化。
supportedOptions
我們再來看下NetworkChannel的其他方法實現，首先來看supportedOptions：
//sun.nio.ch.ServerSocketChannelImpl#supportedOptions
@Override
public final Set<SocketOption<?>> supportedOptions() {
return DefaultOptionsHolder.defaultOptions;
}
//sun.nio.ch.ServerSocketChannelImpl.DefaultOptionsHolder
private static class DefaultOptionsHolder {
static final Set<SocketOption<?>> defaultOptions = defaultOptions();
private static Set<SocketOption<?>> defaultOptions() {
HashSet<SocketOption<?>> set = new HashSet<>();
set.add(StandardSocketOptions.SO_RCVBUF);
set.add(StandardSocketOptions.SO_REUSEADDR);
if (Net.isReusePortAvailable()) {
set.add(StandardSocketOptions.SO_REUSEPORT);
}
set.add(StandardSocketOptions.IP_TOS);
set.addAll(ExtendedSocketOptions.options(SOCK_STREAM));
//返回不可修改的HashSet
return Collections.unmodifiableSet(set);
}
}
複製程式碼對上述配置中的一些配置我們大致來瞅眼:
//java.net.StandardSocketOptions
//socket接受快取大小
public static final SocketOption<Integer> SO_RCVBUF =
new StdSocketOption<Integer>("SO_RCVBUF", Integer.class);
//是否可重用地址
public static final SocketOption<Boolean> SO_REUSEADDR =
new StdSocketOption<Boolean>("SO_REUSEADDR", Boolean.class);
//是否可重用port
public static final SocketOption<Boolean> SO_REUSEPORT =
new StdSocketOption<Boolean>("SO_REUSEPORT", Boolean.class);
//Internet協議（IP）標頭（header）中的服務型別（ToS）。
public static final SocketOption<Integer> IP_TOS =
new StdSocketOption<Integer>("IP_TOS", Integer.class);
複製程式碼setOption實現
知道了上面的支援配置，我們來看下setOption實現細節:
//sun.nio.ch.ServerSocketChannelImpl#setOption
@Override
public <T> ServerSocketChannel setOption(SocketOption<T> name, T value)
throws IOException
{
Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
ensureOpen();
if (name == StandardSocketOptions.IP_TOS) {
ProtocolFamily family = Net.isIPv6Available() ?
StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
Net.setSocketOption(fd, family, name, value);
return this;
}
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
isReuseAddress = (Boolean)value;
} else {
// no options that require special handling
Net.setSocketOption(fd, Net.UNSPEC, name, value);
}
return this;
}
}
複製程式碼這裡，大家就能看到supportedOptions().contains(name)的作用了，首先會進行支援配置的判斷，然後進行正常的設定邏輯。裡面對於Socket配置設定主要執行了Net.setSocketOption，這裡，就只對其程式碼做中文註釋就好，整個邏輯過程沒有太複雜的。
static void setSocketOption(FileDescriptor fd, ProtocolFamily family,
SocketOption<?> name, Object value)
throws IOException
{
if (value == null)
throw new IllegalArgumentException("Invalid option value");
// only simple values supported by this method
Class<?> type = name.type();
if (extendedOptions.isOptionSupported(name)) {
extendedOptions.setOption(fd, name, value);
return;
}
//非整形和布林型，則丟擲斷言錯誤
if (type != Integer.class && type != Boolean.class)
throw new AssertionError("Should not reach here");
// special handling
if (name == StandardSocketOptions.SO_RCVBUF ||
name == StandardSocketOptions.SO_SNDBUF)
{
//判斷接受和傳送緩衝區大小
int i = ((Integer)value).intValue();
if (i < 0)
throw new IllegalArgumentException("Invalid send/receive buffer size");
}
//緩衝區有資料，延遲關閉socket的的時間
if (name == StandardSocketOptions.SO_LINGER) {
int i = ((Integer)value).intValue();
if (i < 0)
value = Integer.valueOf(-1);
if (i > 65535)
value = Integer.valueOf(65535);
}
//UDP單播
if (name == StandardSocketOptions.IP_TOS) {
int i = ((Integer)value).intValue();
if (i < 0 || i > 255)
throw new IllegalArgumentException("Invalid IP_TOS value");
}
//UDP多播
if (name == StandardSocketOptions.IP_MULTICAST_TTL) {
int i = ((Integer)value).intValue();
if (i < 0 || i > 255)
throw new IllegalArgumentException("Invalid TTL/hop value");
}
// map option name to platform level/name
OptionKey key = SocketOptionRegistry.findOption(name, family);
if (key == null)
throw new AssertionError("Option not found");
int arg;
//轉換配置引數值
if (type == Integer.class) {
arg = ((Integer)value).intValue();
} else {
boolean b = ((Boolean)value).booleanValue();
arg = (b) ? 1 : 0;
}
boolean mayNeedConversion = (family == UNSPEC);
boolean isIPv6 = (family == StandardProtocolFamily.INET6);
//設定檔案描述符的值及其他
setIntOption0(fd, mayNeedConversion, key.level(), key.name(), arg, isIPv6);
}
複製程式碼getOption
接下來，我們來看getOption實現，原始碼如下:
//sun.nio.ch.ServerSocketChannelImpl#getOption
@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SocketOption<T> name)
throws IOException
{
Objects.requireNonNull(name);
//非通道支援選項，則丟擲UnsupportedOperationException
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
ensureOpen();
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
return (T)Boolean.valueOf(isReuseAddress);
}
//假如獲取的不是上面的配置，則委託給Net來處理
// no options that require special handling
return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
}
}
//sun.nio.ch.Net#getSocketOption
static Object getSocketOption(FileDescriptor fd, ProtocolFamily family,
SocketOption<?> name)
throws IOException
{
Class<?> type = name.type();
if (extendedOptions.isOptionSupported(name)) {
return extendedOptions.getOption(fd, name);
}
//只支援整形和布林型，否則丟擲斷言錯誤
// only simple values supported by this method
if (type != Integer.class && type != Boolean.class)
throw new AssertionError("Should not reach here");
// map option name to platform level/name
OptionKey key = SocketOptionRegistry.findOption(name, family);
if (key == null)
throw new AssertionError("Option not found");
boolean mayNeedConversion = (family == UNSPEC);
//獲取檔案描述的選項配置
int value = getIntOption0(fd, mayNeedConversion, key.level(), key.name());
if (type == Integer.class) {
return Integer.valueOf(value);
} else {
//我們要看到前面支援配置處的原始碼其支援的型別要麼是Boolean，要麼是Integer
//所以，返回值為Boolean.FALSE 或 Boolean.TRUE也就不足為奇了
return (value == 0) ? Boolean.FALSE : Boolean.TRUE;
}
}
複製程式碼ServerSocketChannel與ServerSocket在bind處的異同
在Net.bind一節中，我們最後說了一個注意點，每個連線過來的時候都會建立一個Socket來供此連線進行操作，這個在accept方法中可以看到，其在得到連線之後，就 new SocketChannelImpl(provider(), newfd, isa)這個物件。那這裡，就引出一個話題，我們在使用bind方法的時候，是不是也應該繫結到一個Socket之上呢，那之前bio是怎麼做呢，我們先來回顧一下。 我們之前在呼叫java.net.ServerSocket#ServerSocket(int, int, java.net.InetAddress)方法的時候，裡面有一個setImpl():
//java.net.ServerSocket
public ServerSocket(int port, int backlog, InetAddress bindAddr) throws IOException {
setImpl();
if (port < 0 || port > 0xFFFF)
throw new IllegalArgumentException(
"Port value out of range: " + port);
if (backlog < 1)
backlog = 50;
try {
bind(new InetSocketAddress(bindAddr, port), backlog);
} catch(SecurityException e) {
close();
throw e;
} catch(IOException e) {
close();
throw e;
}
}
//java.net.ServerSocket#setImpl
private void setImpl() {
if (factory != null) {
impl = factory.createSocketImpl();
checkOldImpl();
} else {
// No need to do a checkOldImpl() here, we know it's an up to date
// SocketImpl!
impl = new SocksSocketImpl();
}
if (impl != null)
impl.setServerSocket(this);
}
複製程式碼但是，我們此處的重點在bind(new InetSocketAddress(bindAddr, port), backlog);，這裡的程式碼如下:
//java.net.ServerSocket
public void bind(SocketAddress endpoint, int backlog) throws IOException {
if (isClosed())
throw new SocketException("Socket is closed");
if (!oldImpl && isBound())
throw new SocketException("Already bound");
if (endpoint == null)
endpoint = new InetSocketAddress(0);
if (!(endpoint instanceof InetSocketAddress))
throw new IllegalArgumentException("Unsupported address type");
InetSocketAddress epoint = (InetSocketAddress) endpoint;
if (epoint.isUnresolved())
throw new SocketException("Unresolved address");
if (backlog < 1)
backlog = 50;
try {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkListen(epoint.getPort());
//重點！！
getImpl().bind(epoint.getAddress(), epoint.getPort());
getImpl().listen(backlog);
bound = true;
} catch(SecurityException e) {
bound = false;
throw e;
} catch(IOException e) {
bound = false;
throw e;
}
}
複製程式碼我們有看到 getImpl()我標示了重點，這裡面做了什麼，我們走進去:
//java.net.ServerSocket#getImpl
SocketImpl getImpl() throws SocketException {
if (!created)
createImpl();
return impl;
}
複製程式碼在整個過程中created還是物件剛建立時的初始值，為false，那麼，鐵定會進入createImpl()方法中:
//java.net.ServerSocket#createImpl
void createImpl() throws SocketException {
if (impl == null)
setImpl();
try {
impl.create(true);
created = true;
} catch (IOException e) {
throw new SocketException(e.getMessage());
}
}
複製程式碼而此處，因為前面impl已經賦值，所以，會走impl.create(true)，進而將created設定為true。而此刻，終於到我想講的重點了:
//java.net.AbstractPlainSocketImpl#create
protected synchronized void create(boolean stream) throws IOException {
this.stream = stream;
if (!stream) {
ResourceManager.beforeUdpCreate();
// only create the fd after we know we will be able to create the socket
fd = new FileDescriptor();
try {
socketCreate(false);
SocketCleanable.register(fd);
} catch (IOException ioe) {
ResourceManager.afterUdpClose();
fd = null;
throw ioe;
}
} else {
fd = new FileDescriptor();
socketCreate(true);
SocketCleanable.register(fd);
}
if (socket != null)
socket.setCreated();
if (serverSocket != null)
serverSocket.setCreated();
}
複製程式碼可以看到，socketCreate(true);，它的實現如下:
@Override
void socketCreate(boolean stream) throws IOException {
if (fd == null)
throw new SocketException("Socket closed");
int newfd = socket0(stream);
fdAccess.set(fd, newfd);
}
複製程式碼通過本地方法socket0(stream)得到了一個檔案描述符，由此，Socket建立了出來，然後進行相應的繫結。 我們再把眼光放回到sun.nio.ch.ServerSocketChannelImpl#accept()中，這裡new的SocketChannelImpl物件是得到連線之後做的事情，那對於伺服器來講，繫結時候用的Socket呢，這裡，我們在使用ServerSocketChannel的時候，往往要使用JDK給我們提供的對我統一的方法open，也是為了降低我們使用的複雜度，這裡是java.nio.channels.ServerSocketChannel#open:
//java.nio.channels.ServerSocketChannel#open
public static ServerSocketChannel open() throws IOException {
return SelectorProvider.provider().openServerSocketChannel();
}
//sun.nio.ch.SelectorProviderImpl#openServerSocketChannel
public ServerSocketChannel openServerSocketChannel() throws IOException {
return new ServerSocketChannelImpl(this);
}
//sun.nio.ch.ServerSocketChannelImpl#ServerSocketChannelImpl(SelectorProvider)
ServerSocketChannelImpl(SelectorProvider sp) throws IOException {
super(sp);
this.fd = Net.serverSocket(true);
this.fdVal = IOUtil.fdVal(fd);
}
//sun.nio.ch.Net#serverSocket
static FileDescriptor serverSocket(boolean stream) {
return IOUtil.newFD(socket0(isIPv6Available(), stream, true, fastLoopback));
}
複製程式碼可以看到，只要new了一個ServerSocketChannelImpl物件，就相當於拿到了一個socket然後bind也就有著落了。但是，我們要注意下細節ServerSocketChannel#open得到的是ServerSocketChannel型別。我們accept到一個客戶端來的連線後，應該在客戶端與伺服器之間建立一個Socket通道來供兩者通訊操作的，所以，sun.nio.ch.ServerSocketChannelImpl#accept()中所做的是SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);，得到的是SocketChannel型別的物件，這樣，就可以將Socket的讀寫資料的方法定義在這個類裡面。
由ServerSocketChannel的socket方法延伸的
關於ServerSocketChannel，我們還有方法需要接觸一下，如socket():
//sun.nio.ch.ServerSocketChannelImpl#socket
@Override
public ServerSocket socket() {
synchronized (stateLock) {
if (socket == null)
socket = ServerSocketAdaptor.create(this);
return socket;
}
}
複製程式碼我們看到了ServerSocketAdaptor，我們通過此類的註釋可知，這是一個和ServerSocket呼叫一樣，但是底層是用ServerSocketChannelImpl來實現的一個類，其適配是的目的是適配我們使用ServerSocket的方式，所以該ServerSocketAdaptor繼承ServerSocket並按順序重寫了它的方法，所以，我們在寫這塊兒程式碼的時候也就有了新的選擇。
InterruptibleChannel 與可中斷 IO這一篇文章中已經涉及過java.nio.channels.spi.AbstractInterruptibleChannel#close的實現，這裡，我們再來回顧下其中的某些細節，順帶引出我們新的話題:
//java.nio.channels.spi.AbstractInterruptibleChannel#close
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
}
}
}
}
//sun.nio.ch.ServerSocketChannelImpl#implCloseSelectableChannel
@Override
protected void implCloseSelectableChannel() throws IOException {
assert !isOpen();
boolean interrupted = false;
boolean blocking;
// set state to ST_CLOSING
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
blocking = isBlocking();
}
// wait for any outstanding accept to complete
if (blocking) {
synchronized (stateLock) {
assert state == ST_CLOSING;
long th = thread;
if (th != 0) {
//本地執行緒不為null，則本地Socket預先關閉
//並通知執行緒通知關閉
nd.preClose(fd);
NativeThread.signal(th);
// wait for accept operation to end
while (thread != 0) {
try {
stateLock.wait();
} catch (InterruptedException e) {
interrupted = true;
}
}
}
}
} else {
// non-blocking mode: wait for accept to complete
acceptLock.lock();
acceptLock.unlock();
}
// set state to ST_KILLPENDING
synchronized (stateLock) {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
}
// close socket if not registered with Selector
//如果未在Selector上註冊，直接kill掉
//即關閉檔案描述
if (!isRegistered())
kill();
// restore interrupt status
//印證了我們上一篇中在非同步打斷中若是通過執行緒的中斷方法中斷執行緒的話
//最後要設定該執行緒狀態是interrupt
if (interrupted)
Thread.currentThread().interrupt();
}
@Override
public void kill() throws IOException {
synchronized (stateLock) {
if (state == ST_KILLPENDING) {
state = ST_KILLED;
nd.close(fd);
}
}
}
複製程式碼channel的close()應用
也是因為close()並沒有在InterruptibleChannel 與可中斷 IO這一篇文章中進行具體的講解應用，這裡其應用的更多是在SocketChannel這裡，其更多的涉及到客戶端與服務端建立連線交換資料，所以斷開連線後，將不用的Channel關閉是很正常的。 這裡，在sun.nio.ch.ServerSocketChannelImpl#accept()中的原始碼中:
@Override
public SocketChannel accept() throws IOException {
...
// newly accepted socket is initially in blocking mode
IOUtil.configureBlocking(newfd, true);
InetSocketAddress isa = isaa[0];
SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
// check permitted to accept connections from the remote address
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
} catch (SecurityException x) {
sc.close();
throw x;
}
}
return sc;
} finally {
acceptLock.unlock();
}
}
複製程式碼這裡通過對所接收的連線的遠端地址做合法性判斷，假如驗證出現異常，則關閉上面建立的SocketChannel。 還有一個關於close()的實際用法，在客戶端建立連線的時候，如果連接出異常，同樣是要關閉所建立的Socket:
//java.nio.channels.SocketChannel#open(java.net.SocketAddress)
public static SocketChannel open(SocketAddress remote)
throws IOException
{
SocketChannel sc = open();
try {
sc.connect(remote);
} catch (Throwable x) {
try {
sc.close();
} catch (Throwable suppressed) {
x.addSuppressed(suppressed);
}
throw x;
}
assert sc.isConnected();
return sc;
}
複製程式碼接著，我們在implCloseSelectableChannel中會發現nd.preClose(fd);與nd.close(fd);，這個在SocketChannelImpl與ServerSocketChannelImpl兩者對於implCloseSelectableChannel實現中都可以看到，這個nd是什麼，這裡，我們拿ServerSocketChannelImpl來講，在這個類的最後面有一段靜態程式碼塊(SocketChannelImpl同理)，也就是在這個類載入的時候就會執行:
//C:/Program Files/Java/jdk-11.0.1/lib/src.zip!/java.base/sun/nio/ch/ServerSocketChannelImpl.java:550
static {
//載入nio，net資源庫
IOUtil.load();
initIDs();
nd = new SocketDispatcher();
}
複製程式碼也就是說，在ServerSocketChannelImpl這個類位元組碼載入的時候，就會建立SocketDispatcher物件。通過SocketDispatcher允許在不同的平臺呼叫不同的本地方法進行讀寫操作,然後基於這個類，我們就可以在sun.nio.ch.SocketChannelImpl做Socket的I/O操作。
//sun.nio.ch.SocketDispatcher
class SocketDispatcher extends NativeDispatcher
{
static {
IOUtil.load();
}
//讀操作
int read(FileDescriptor fd, long address, int len) throws IOException {
return read0(fd, address, len);
}
long readv(FileDescriptor fd, long address, int len) throws IOException {
return readv0(fd, address, len);
}
//寫操作
int write(FileDescriptor fd, long address, int len) throws IOException {
return write0(fd, address, len);
}
long writev(FileDescriptor fd, long address, int len) throws IOException {
return writev0(fd, address, len);
}
//預關閉檔案描述符
void preClose(FileDescriptor fd) throws IOException {
preClose0(fd);
}
//關閉檔案描述
void close(FileDescriptor fd) throws IOException {
close0(fd);
}
//-- Native methods
static native int read0(FileDescriptor fd, long address, int len)
throws IOException;
static native long readv0(FileDescriptor fd, long address, int len)
throws IOException;
static native int write0(FileDescriptor fd, long address, int len)
throws IOException;
static native long writev0