BIO到NIO源碼的一些事兒之NIO 中

收藏待读

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

本系列源碼解讀基於JDK11 api細節可能與其他版本有所差別,請自行解決jdk版本問題。

Channel解讀

接上一篇 BIO到NIO源碼的一些事兒之NIO 上

賦予Channel支持網絡socket的能力

我們最初的目的就是為了增強Socket,基於這個基本需求,沒有條件創造條件,於是為了讓Channel擁有網絡socket的能力,這裡定義了一個 java.nio.channels.NetworkChannel 接口。花不多說,我們來看這個接口的定義:

public interface NetworkChannel extends Channel
{
    NetworkChannel bind(SocketAddress local) throws IOException;

    SocketAddress getLocalAddress() throws IOException;

     NetworkChannel setOption(SocketOption name, T value) throws IOException;

     T getOption(SocketOption name) throws IOException;

    Set<SocketOption> supportedOptions();
}

通過 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.listennative 方法,源碼如下:

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工具類的 bindlisten ,完成實際的 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 SO_RCVBUF =
        new StdSocketOption("SO_RCVBUF", Integer.class);
//是否可重用地址  
public static final SocketOption SO_REUSEADDR =
        new StdSocketOption("SO_REUSEADDR", Boolean.class);
//是否可重用port
public static final SocketOption SO_REUSEPORT =
        new StdSocketOption("SO_REUSEPORT", Boolean.class);
//Internet協議(IP)標頭(header)中的服務類型(ToS)。
public static final SocketOption IP_TOS =
        new StdSocketOption("IP_TOS", Integer.class);

setOption實現

知道了上面的支持配置,我們來看下 setOption 實現細節:

//sun.nio.ch.ServerSocketChannelImpl#setOption
@Override
public  ServerSocketChannel setOption(SocketOption 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  65535)
            value = Integer.valueOf(65535);
    }
    //UDP單播  
    if (name == StandardSocketOptions.IP_TOS) {
        int i = ((Integer)value).intValue();
        if (i  255)
            throw new IllegalArgumentException("Invalid IP_TOS value");
    }
    //UDP多播  
    if (name == StandardSocketOptions.IP_MULTICAST_TTL) {
        int i = ((Integer)value).intValue();
        if (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 getOption(SocketOption 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  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); ,這個在 SocketChannelImplServerSocketChannelImpl 兩者對於 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(FileDescriptor fd, long address, int len)
        throws IOException;

    static native void preClose0(FileDescriptor fd) throws IOException;

    static native void close0(FileDescriptor fd) throws IOException;
}

FileDescriptor

我們有看到 FileDescriptor 在前面代碼中有大量的出現,這裡,我們對它來專門介紹。通過FileDescriptor 這個類的實例來充當底層機器特定結構的不透明處理,表示打開文件,打開socket或其他位元組源或接收器。

文件描述符的主要用途是創建一個 FileInputStream或 FileOutputStream來包含它。

注意: 應用程序不應創建自己的文件描述符。

我們來看其部分源碼:

public final class FileDescriptor {

    private int fd;

    private long handle;

    private Closeable parent;
    private List otherParents;
    private boolean closed;

    /**
     * true, if file is opened for appending.
     */
    private boolean append;

    static {
        initIDs();
    }
    /**
     * 在未明確關閉FileDescriptor的情況下進行清理.
     */
    private PhantomCleanable cleanup;

    /**
     * 構造一個無效的FileDescriptor對象,fd或handle會在之後進行設定
     */
    public FileDescriptor() {
        fd = -1;
        handle = -1;
    }

    /**
     * Used for standard input, output, and error only.
     * For Windows the corresponding handle is initialized.
     * For Unix the append mode is cached.
     * 僅用於標準輸入,輸出和錯誤。
     * 對於Windows,初始化相應的句柄。
     * 對於Unix,緩存附加模式。
     * @param fd the raw fd number (0, 1, 2)
     */
    private FileDescriptor(int fd) {
        this.fd = fd;
        this.handle = getHandle(fd);
        this.append = getAppend(fd);
    }
    ...
}

我們平時所用的標準輸入,輸出,錯誤流的句柄可以如下,通常,我們不會直接使用它們,而是使用 java.lang.System.injava.lang.System#outjava.lang.System#err :

public static final FileDescriptor in = new FileDescriptor(0);
public static final FileDescriptor out = new FileDescriptor(1);
public static final FileDescriptor err = new FileDescriptor(2);

測試該文件描述符是否有效可以使用如下方法:

//java.io.FileDescriptor#valid
public boolean valid() {
        return (handle != -1) || (fd != -1);
    }

返回值為true的話,那麼這個文件描述符對象所代表的 socket 文件操作 或其他活動的網絡連接都是有效的,反之,false則是無效。

更多內容,讀者可以自行深入源碼,此處就不過多解釋了。為了讓大家可以更好的理解上述內容,我們會在後面的部分還要進一步涉及一下。

NIO包下SocketChannel解讀

在前面,我們已經接觸了 SocketChannel ,這裡,來接觸下細節。

同樣,我們也可以通過調用此類的 open 方法來創建 socket channel 。這裡需要注意:

  • 無法為任意預先存在的 socket 創建 channel
  • 新創建的 socket channel 已打開但尚未連接。
  • 嘗試在未連接的 channel 上調用 I/O 操作將導致拋出 NotYetConnectedException
  • 可以通過調用 connect 方法連接 socket channel ;
  • 一旦連接後, socket channel 會保持連接狀態,直到它關閉。
  • 是否有連接 socket channel 可以通過確定調用其 isConnected 方法。

socket channel 支持 非阻塞連接:

  • 可以先創建 socket channel ,然後可以通過 connect 方法建立到遠程 socket 的連接。
  • 通過調用 finishConnect 方法來結束連接。
  • 判斷是否正在進行連接操作可以通過調用 isConnectionPending 方法來確定。

socket channel 支持異步關閉,類似於 Channel 類中的異步關閉操作。

  • 如果 socket 的輸入端被一個線程關閉而另一個線程在此 socket channel 上因在進行讀操作而被阻塞,那麼被阻塞線程中的讀操作將不讀取任何位元組並將返回 -1
  • 如果 socket 的輸出端被一個線程關閉而另一個線程在 socket channel 上因在進行寫操作而被阻塞,則被阻塞的線程將收到 AsynchronousCloseException

接下來,我們來看其具體實現方法。

ServerSocketChannel與SocketChannel的open()

//java.nio.channels.SocketChannel#open()
public static SocketChannel open() throws IOException {
    return SelectorProvider.provider().openSocketChannel();
}
//java.nio.channels.SocketChannel#open(java.net.SocketAddress)
//這個方法省的我們再次調用connect了
public static SocketChannel open(SocketAddress remote)
    throws IOException
{
    //默認是堵塞的,這個在AbstractSelectableChannel處討論過了
    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;
}
//sun.nio.ch.SelectorProviderImpl#openSocketChannel
public SocketChannel openSocketChannel() throws IOException {
    return new SocketChannelImpl(this);
}
//sun.nio.ch.SocketChannelImpl#SocketChannelImpl(java.nio.channels.spi.SelectorProvider)
SocketChannelImpl(SelectorProvider sp) throws IOException {
    super(sp);
     //調用socket函數,true表示TCP
    this.fd = Net.socket(true);
    this.fdVal = IOUtil.fdVal(fd);
}
//sun.nio.ch.Net#socket(boolean)
static FileDescriptor socket(boolean stream) throws IOException {
    return socket(UNSPEC, stream);
}
//sun.nio.ch.Net#socket(java.net.ProtocolFamily, boolean)
static FileDescriptor socket(ProtocolFamily family, boolean stream)
    throws IOException {
    boolean preferIPv6 = isIPv6Available() &&
        (family != StandardProtocolFamily.INET);
    return IOUtil.newFD(socket0(preferIPv6, stream, false, fastLoopback));
}
//sun.nio.ch.IOUtil#newFD
public static FileDescriptor newFD(int i) {
    FileDescriptor fd = new FileDescriptor();
    setfdVal(fd, i);
    return fd;
}
static native void setfdVal(FileDescriptor fd, int value);

關於 Net.socket(true) ,我們前面已經提到過了,這裡,通過其底層源碼來再次調教下 (此處不想看可以跳過):

JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_socket0(JNIEnv *env, jclass cl, jboolean preferIPv6,
                            jboolean stream, jboolean reuse, jboolean ignored)
{
    int fd;
    //位元組流還是數據報,TCP對應SOCK_STREAM,UDP對應SOCK_DGRAM,此處傳入的stream=true;
    int type = (stream ? SOCK_STREAM : SOCK_DGRAM);
    //判斷是IPV6還是IPV4
    int domain = (ipv6_available() && preferIPv6) ? AF_INET6 : AF_INET;

    //調用Linux的socket函數,domain為代表協議;
    //type為套接字類型,protocol設置為0來表示使用默認的傳輸協議
    fd = socket(domain, type, 0);
    //出錯
    if (fd < 0) {
        return handleSocketError(env, errno);
    }

    /* Disable IPV6_V6ONLY to ensure dual-socket support */
    if (domain == AF_INET6) {
        int arg = 0;
        //arg=1設置ipv6的socket只接收ipv6地址的報文,arg=0表示也可接受ipv4的請求
        if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&arg,
                       sizeof(int)) < 0) {
            JNU_ThrowByNameWithLastError(env,
                                         JNU_JAVANETPKG "SocketException",
                                         "Unable to set IPV6_V6ONLY");
            close(fd);
            return -1;
        }
    }

    //SO_REUSEADDR有四種用途:
    //1.當有一個有相同本地地址和端口的socket1處於TIME_WAIT狀態時,而你啟動的程序的socket2要佔用該地址和端口,你的程序就要用到該選項。 
    //2.SO_REUSEADDR允許同一port上啟動同一服務器的多個實例(多個進程)。但每個實例綁定的IP地址是不能相同的。
    //3.SO_REUSEADDR允許單個進程綁定相同的端口到多個socket上,但每個socket綁定的ip地址不同。 
   //4.SO_REUSEADDR允許完全相同的地址和端口的重複綁定。但這只用於UDP的多播,不用於TCP;
    if (reuse) {
        int arg = 1;
        if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&arg,
                       sizeof(arg)) < 0) {
            JNU_ThrowByNameWithLastError(env,
                                         JNU_JAVANETPKG "SocketException",
                                         "Unable to set SO_REUSEADDR");
            close(fd);
            return -1;
        }
    }

#if defined(__linux__)
    if (type == SOCK_DGRAM) {
        int arg = 0;
        int level = (domain == AF_INET6) ? IPPROTO_IPV6 : IPPROTO_IP;
        if ((setsockopt(fd, level, IP_MULTICAST_ALL, (char*)&arg, sizeof(arg)) < 0) &&
            (errno != ENOPROTOOPT)) {
            JNU_ThrowByNameWithLastError(env,
                                         JNU_JAVANETPKG "SocketException",
                                         "Unable to set IP_MULTICAST_ALL");
            close(fd);
            return -1;
        }
    }

     //IPV6_MULTICAST_HOPS用於控制多播的範圍,
     // 1表示只在本地網絡轉發,
     //更多介紹請參考(http://www.ctt.sbras.ru/cgi-bin/www/unix_help/unix-man?ip6+4);
    /* By default, Linux uses the route default */
    if (domain == AF_INET6 && type == SOCK_DGRAM) {
        int arg = 1;
        if (setsockopt(fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &arg,
                       sizeof(arg)) < 0) {
            JNU_ThrowByNameWithLastError(env,
                                         JNU_JAVANETPKG "SocketException",
                                         "Unable to set IPV6_MULTICAST_HOPS");
            close(fd);
            return -1;
        }
    }
#endif
    return fd;
}

Linux 3.9之後加入了 SO_REUSEPORT 配置,這個配置很強大,多個 socket (不管是處於監聽還是非監聽,不管是TCP還是UDP)只要在綁定之前設置了 SO_REUSEPORT 屬性,那麼就可以綁定到完全相同的地址和端口。

為了阻止」port 劫持」( Port hijacking )有一個特別的限制:所有希望共享源地址和端口的socket都必須擁有相同的有效用戶id( effective user ID )。這樣一個用戶就不能從另一個用戶那裡」偷取」端口。另外,內核在處理 SO_REUSEPORT socket 的時候使用了其它系統上沒有用到的」特殊技巧」:

  • 對於UDP socket,內核嘗試平均的轉發數據報;
  • 對於TCP監聽socket,內核嘗試將新的客戶連接請求(由accept返回)平均的交給共享同一地址和端口的socket(服務器監聽socket)。

例如:一個簡單的服務器程序的多個實例可以使用 SO_REUSEPORT socket ,這樣就實現一個簡單的負載均衡,因為內核已經把請求的分配都做了。

在前面的代碼中可以看到,在這個 socket 創建成功之後,調用 IOUtil.newFD 創建了文件描述符

。這裡,我只是想知道這個Socket是可以輸入呢,還是可以讀呢,還是有錯呢,參考 FileDescriptor 這一節最後那幾個標準狀態的設定,其實這裡也是一樣,因為我們要往Socket中寫和讀,其標準狀態無非就這三種:輸入,輸出,出錯。而這個Socket是綁定在 SocketChannel 上的,那就把 FileDescriptor 也綁定到上面即可,這樣我們就可以獲取到它的狀態了。由於FileDescriptor沒有提供外部設置fd的方法,setfdVal是通過本地方法實現的:

JNIEXPORT void JNICALL
Java_sun_nio_ch_IOUtil_setfdVal(JNIEnv *env, jclass clazz, jobject fdo, jint val)
{
    (*env)->SetIntField(env, fdo, fd_fdID, val);
}

假如各位有對Linux下的shell編程或者命令有了解的話,我們知道,shell對報錯進行重定向要使用2>,也就是將錯誤信息由2號所指向的通道寫出,這裡0和1 同樣指向一個通道。此處同樣也代表了狀態,這樣就可以對代表Socket的狀態進行操作了,也就是改變 SelectionKeyinterest ops ,即首先對 SelectionKey 按輸入輸出類型進行分類,然後我們的讀寫狀態的操作也就有着落了。此處我們打個戳,在下一篇中會對其進行細節講解。

我們回歸到 SocketChannelopen 方法中。我們可以看到, SelectorProvider.provider().openSocketChannel() 返回的是 SocketChannelImpl 對象實例。在 SocketChannelImpl(SelectorProvider sp) 中我們並未看到其對 this.state 進行值操作,也就是其默認為0,即 ST_UNCONNECTED (未連接狀態),同時Socket默認是堵塞的。

所以,一般情況下,當採用異步方式時,使用不帶參數的open方法比較常見,這樣,我們會隨之調用 configureBlocking 來設置非堵塞。

SocketChannel的connect解讀

由前面可知,我們調用 connect 方法連接到遠程服務器,其源碼如下:

//sun.nio.ch.SocketChannelImpl#connect
@Override
public boolean connect(SocketAddress sa) throws IOException {
    InetSocketAddress isa = Net.checkAddress(sa);
    SecurityManager sm = System.getSecurityManager();
    if (sm != null)
        sm.checkConnect(isa.getAddress().getHostAddress(), isa.getPort());

    InetAddress ia = isa.getAddress();
    if (ia.isAnyLocalAddress())
        ia = InetAddress.getLocalHost();

    try {
        readLock.lock();
        try {
            writeLock.lock();
            try {
                int n = 0;
                boolean blocking = isBlocking();
                try {
                    //支持線程中斷,通過設置當前線程的Interruptible blocker屬性實現
                    beginConnect(blocking, isa);
                    do {
                    //調用connect函數實現,如果採用堵塞模式,會一直等待,直到成功或出//現異常
                        n = Net.connect(fd, ia, isa.getPort());
                    } while (n == IOStatus.INTERRUPTED && isOpen());
                } finally {
                    endConnect(blocking, (n > 0));
                }
                assert IOStatus.check(n);
                //連接成功
                return n > 0;
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
    } catch (IOException ioe) {
        // connect failed, close the channel
        close();
        throw SocketExceptions.of(ioe, isa);
    }
}

關於 beginConnectendConnect ,是針對 AbstractInterruptibleChannelbegin()end 方法的一種增強。這裡我們需要知道的是,假如是非阻塞Channel的話,我們無須去關心連接過程的打斷。顧名思義,只有阻塞等待才需要去考慮打斷這一場景的出現。剩下的細節我已經在代碼中進行了完整的注釋,讀者可自行查看。

//sun.nio.ch.SocketChannelImpl#beginConnect
private void beginConnect(boolean blocking, InetSocketAddress isa)
    throws IOException
{   //只有阻塞的時候才會進入begin
    if (blocking) {
        // set hook for Thread.interrupt
        //支持線程中斷,通過設置當前線程的Interruptible blocker屬性實現
        begin();
    }
    synchronized (stateLock) {
        //默認為open, 除非調用了close方法
        ensureOpen();
        //檢查連接狀態
        int state = this.state;
        if (state == ST_CONNECTED)
            throw new AlreadyConnectedException();
        if (state == ST_CONNECTIONPENDING)
            throw new ConnectionPendingException();
        //斷言當前的狀態是否是未連接狀態,如果是,賦值表示正在連接中
        assert state == ST_UNCONNECTED;
        //表示正在連接中
        this.state = ST_CONNECTIONPENDING;
        //只有未綁定本地地址也就是說未調用bind方法才執行,
        //該方法在ServerSocketChannel中也見過
        if (localAddress == null)
            NetHooks.beforeTcpConnect(fd, isa.getAddress(), isa.getPort());
        remoteAddress = isa;

        if (blocking) {
            // record thread so it can be signalled if needed
            readerThread = NativeThread.current();
        }
    }
}

在連接過程中,我們需要注意的就是幾個連接的狀態: ST_UNCONNECTEDST_CONNECTEDST_CONNECTIONPENDINGST_CLOSINGST_KILLPENDINGST_KILLED ,也是因為其是一個公共狀態,可能會有多個線程對其進行連接操作的。所以, state 被定義為一個 volatile 變量,這個變量在改變的時候需要有 stateLock 這個對象來作為 synchronized 鎖對象來控制同步操作的。

//sun.nio.ch.SocketChannelImpl#endConnect
private void endConnect(boolean blocking, boolean completed)
    throws IOException
{
    endRead(blocking, completed);
    //當上面代碼中n>0,說明連接成功,更新狀態為ST_CONNECTED
    if (completed) {
        synchronized (stateLock) {
            if (state == ST_CONNECTIONPENDING) {
                localAddress = Net.localAddress(fd);
                state = ST_CONNECTED;
            }
        }
    }
}
//sun.nio.ch.SocketChannelImpl#endRead
private void endRead(boolean blocking, boolean completed)
    throws AsynchronousCloseException
{   //當阻塞狀態下的話,才進入
    if (blocking) {
        synchronized (stateLock) {
            readerThread = 0;
            // notify any thread waiting in implCloseSelectableChannel
            if (state == ST_CLOSING) {
                stateLock.notifyAll();
            }
        }
        //和begin成對出現,當線程中斷時,拋出ClosedByInterruptException
        // remove hook for Thread.interrupt
        end(completed);
    }
}

我們來關注 connect 中的 Net.connect(fd, ia, isa.getPort()) 方法:

//sun.nio.ch.Net#connect
static int connect(FileDescriptor fd, InetAddress remote, int remotePort)
    throws IOException
{
    return connect(UNSPEC, fd, remote, remotePort);
}
//sun.nio.ch.Net#connect
static int connect(ProtocolFamily family, FileDescriptor fd, InetAddress remote, int remotePort)
    throws IOException
{
    boolean preferIPv6 = isIPv6Available() &&
        (family != StandardProtocolFamily.INET);
    return connect0(preferIPv6, fd, remote, remotePort);
}

該方法最終會調用native方法,具體注釋如下:

JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_connect0(JNIEnv *env, jclass clazz, jboolean preferIPv6,
                             jobject fdo, jobject iao, jint port)
{
    SOCKETADDRESS sa;
    int sa_len = 0;
    int rv;
    //地址轉換為struct sockaddr格式
    if (NET_InetAddressToSockaddr(env, iao, port, &sa, &sa_len, preferIPv6) != 0) {
        return IOS_THROWN;
    }
    //傳入fd和sockaddr,與遠程服務器建立連接,一般就是TCP三次握手
   //如果設置了configureBlocking(false),不會堵塞,否則會堵塞一直到超時或出現異常
    rv = connect(fdval(env, fdo), &sa.sa, sa_len);
    //0表示連接成功,失敗時通過errno獲取具體原因
    if (rv != 0) {
        //非堵塞,連接還未建立(-2)
        if (errno == EINPROGRESS) {
            return IOS_UNAVAILABLE;
        } else if (errno == EINTR) {
            //中斷(-3)
            return IOS_INTERRUPTED;
        }
        return handleSocketError(env, errno);
    }
    //連接建立,一般TCP連接連接都需要時間,因此除非是本地網絡,
    //一般情況下非堵塞模式返回IOS_UNAVAILABLE比較多;
    return 1;
}

從上面可以通過注釋看到,如果是非堵塞,而且連接也並未立馬建立成功,其返回的是-2,也就是連接未建立成功,由之前 beginConnect 部分源碼可知,此時狀態為 ST_CONNECTIONPENDING ,那麼,非阻塞條件下,什麼時候會變為 ST_CONNECTED ?有什麼方法可以查詢狀態或者等待連接完成?

那就讓我們來關注下 sun.nio.ch.SocketChannelImpl#finishConnect

SocketChannelImpl中finishConnect解讀

首先,我們回顧下,前面我們涉及了 sun.nio.ch.ServerSocketAdaptor 的用法,方便我們只有Socket編程習慣人群使用,這裡,我們也就可以看到基本的核心實現邏輯,那麼有 ServerSocketAdaptor 就有 SocketAdaptor ,這裡,在BIO的Socket編程中最後也是調用了 connect(address) 操作:

//java.net.Socket#Socket
private Socket(SocketAddress address, SocketAddress localAddr,
                boolean stream) throws IOException {
    setImpl();

    // backward compatibility
    if (address == null)
        throw new NullPointerException();

    try {
        createImpl(stream);
        if (localAddr != null)
            bind(localAddr);
        connect(address);
    } catch (IOException | IllegalArgumentException | SecurityException e) {
        try {
            close();
        } catch (IOException ce) {
            e.addSuppressed(ce);
        }
        throw e;
    }
}

這裡,我們可以調用 java.nio.channels.SocketChannel#open() ,然後調用所得到的 SocketChannel 對象的 socket() 方法,就可以得到 sun.nio.ch.SocketAdaptor 對象實例了。我們來查看 SocketAdaptor 的connect實現:

//sun.nio.ch.SocketAdaptor#connect
public void connect(SocketAddress remote) throws IOException {
    connect(remote, 0);
}

public void connect(SocketAddress remote, int timeout) throws IOException {
    if (remote == null)
        throw new IllegalArgumentException("connect: The address can't be null");
    if (timeout < 0)
        throw new IllegalArgumentException("connect: timeout can't be negative");

    synchronized (sc.blockingLock()) {
        if (!sc.isBlocking())
            throw new IllegalBlockingModeException();

        try {
            //未設定超時則會一直在此等待直到連接或者出現異常
            // no timeout
            if (timeout == 0) {
                sc.connect(remote);
                return;
            }
            //有超時設定,則會將Socket給設定為非阻塞
            // timed connect
            sc.configureBlocking(false);
            try {
                if (sc.connect(remote))
                    return;
            } finally {
                try {
                    sc.configureBlocking(true);
                } catch (ClosedChannelException e) { }
            }

            long timeoutNanos = NANOSECONDS.convert(timeout, MILLISECONDS);
            long to = timeout;
            for (;;) {
                //通過計算超時時間,在允許的時間範圍內無限循環來進行連接,
                //如果超時,則關閉這個Socket
                long startTime = System.nanoTime();
                if (sc.pollConnected(to)) {
                    boolean connected = sc.finishConnect();
                    //看下文解釋
                    assert connected;
                    break;
                }
                timeoutNanos -= System.nanoTime() - startTime;
                if (timeoutNanos <= 0) {
                    try {
                        sc.close();
                    } catch (IOException x) { }
                    throw new SocketTimeoutException();
                }
                to = MILLISECONDS.convert(timeoutNanos, NANOSECONDS);
            }

        } catch (Exception x) {
            Net.translateException(x, true);
        }
    }

}

這裡先解釋下一個小注意點:在Java中, assert 關鍵字是從 JAVA SE 1.4 引入的,為了避免和 老版本的Java代碼 中使用了 assert 關鍵字導致錯誤, Java在執行的時候默認是不啟動斷言檢查的(這個時候,所有的斷言語句都 將忽略!),如果要開啟斷言檢查,則需要用開關-enableassertions或-ea來開啟。

通過上面的源碼注釋,相信大夥已經知道大致的流程了,那關於 sun.nio.ch.SocketChannelImpl#finishConnect 到底做了什麼,此處,我們來探索一番:

//sun.nio.ch.SocketChannelImpl#finishConnect
@Override
public boolean finishConnect() throws IOException {
    try {
        readLock.lock();
        try {
            writeLock.lock();
            try {
                // no-op if already connected
                if (isConnected())
                    return true;

                boolean blocking = isBlocking();
                boolean connected = false;
                try {
                    beginFinishConnect(blocking);
                    int n = 0;
                    if (blocking) {
                        do {
                            //阻塞情況下,第二個參數傳入true
                            n = checkConnect(fd, true);
                        } while ((n == 0 || n == IOStatus.INTERRUPTED) && isOpen());
                    } else {
                        //非阻塞情況下,第二個參數傳入false
                        n = checkConnect(fd, false);
                    }
                    connected = (n > 0);
                } finally {
                    endFinishConnect(blocking, connected);
                }
                assert (blocking && connected) ^ !blocking;
                return connected;
            } finally {
                writeLock.unlock();
            }
        } finally {
            readLock.unlock();
        }
    } catch (IOException ioe) {
        // connect failed, close the channel
        close();
        throw SocketExceptions.of(ioe, remoteAddress);
    }
}
//sun.nio.ch.SocketChannelImpl#checkConnect
private static native int checkConnect(FileDescriptor fd, boolean block)
    throws IOException;

關於 beginFinishConnectendFinishConnect 和我們之前分析的 sun.nio.ch.SocketChannelImpl#beginConnectsun.nio.ch.SocketChannelImpl#endConnect 過程差不多,不懂讀者可回看。剩下的,就是我們關注的主要核心邏輯 checkConnect(fd, true) ,它也是一個本地方法,涉及到的源碼如下:

JNIEXPORT jint JNICALL
Java_sun_nio_ch_SocketChannelImpl_checkConnect(JNIEnv *env, jobject this,
                                               jobject fdo, jboolean block)
{
    int error = 0;
    socklen_t n = sizeof(int);
    //獲取FileDescriptor中的fd
    jint fd = fdval(env, fdo);
    int result = 0;
    struct pollfd poller;
    //文件描述符
    poller.fd = fd;
    //請求的事件為寫事件
    poller.events = POLLOUT;
    //返回的事件
    poller.revents = 0;
    
    //-1表示阻塞,0表示立即返回,不阻塞進程
    result = poll(&poller, 1, block ? -1 : 0);
    //小於0表示調用失敗
    if (result 0
    if (result > 0) {
        errno = 0;
        result = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &n);
        //出錯
        if (result < 0) {
            return handleSocketError(env, errno);
        //發生錯誤,處理錯誤
        } else if (error) {
            return handleSocketError(env, error);
        } else if ((poller.revents & POLLHUP) != 0) {
            return handleSocketError(env, ENOTCONN);
        }
        //socket已經準備好,可寫,即連接已經建立好
        // connected
        return 1;
    }
    return 0;
}

具體的過程如源碼注釋所示,其中是否阻塞我們在本地方法源碼中和之前 sun.nio.ch.SocketChannelImpl#finishConnect 的行為產生對應。另外,從上面的源碼看到,底層是通過 poll 查詢 socket 的狀態,從而判斷連接是否建立成功;由於在非堵塞模式下, finishConnect 方法會立即返回,根據此處 sun.nio.ch.SocketAdaptor#connect 的處理,其使用循環的方式判斷連接是否建立,在我們的nio編程中,這個是不建議的,屬於半成品,而是建議註冊到 Selector ,通過 ops=OP_CONNECT 獲取連接完成的 SelectionKey ,然後調用 finishConnect 完成連接的建立;

那麼 finishConnect 是否可以不調用呢?答案是否定的,因為只有 finishConnect 中才會將狀態更新為 ST_CONNECTED ,而在調用 readwrite 時都會對狀態進行判斷。

這裡,我們算是引出了我們即將要涉及的 SelectorSelectionKey ,我們會在下一篇中進行詳細講解。

原文 : 一葉知秋

相關閱讀

免责声明:本文内容来源于一葉知秋,已注明原文出处和链接,文章观点不代表立场,如若侵犯到您的权益,或涉不实谣言,敬请向我们提出检举。