【死磕 Java NIO】— 探索 SocketChannel 的核心原理

前两篇文章我们分析了 Channel 及 FileChannel，这篇文章我们探究 SocketChannel的核心原理，毕竟下一个系列就是 **【死磕 Netty】**了。

聊聊Socket

要想掌握 SocketChannel，我们就必须先了解什么是 Socket。要想解释清楚 Socket，就需要了解下 TCP/IP。

注：本文重点在 SocketChannel，所以对 TCP和 Socket仅仅只做相关介绍，有兴趣的同学，麻烦自查专业资料。

TCP/IP 体系结构

学过计算机网络的小伙伴知道，计算机网络是分层的，每层专注于一类事情。OSI 网路模型分为七层，如下：

OSI 模型是理论中的模型，在实际应用中我们使用的是 TCP/IP 四层模型，它对OSI模型重新进行了划分和规整，如下：

网络层次划分清楚了，那怎么传输数据呢？如下图：

计算机A首先在应用层将要发送的数据准备好，然后给传输层， 传输层的主要作用就是为发送端和接收端提供可靠的连接服务，传输层将数据处理完成后给网络层， 网络层的一个核心功能就是数据传输路径的选择。计算机A到计算机B有很多条路，网络层的作用就是负责管理下一步数据应该到那个路由器，选择好路径后，数据就到了网络接入层，该层主要负责将数据从一个路由器发送到另一个路由器。

上图是一个非常清晰的传输过程。但是我们思考两个个问题：

1. 计算机A是怎么知道计算机B的具体位置的呢？
2. 它又怎么知道将该数据包发送给哪个应用程序呢？

TCP/IP协议族已经帮我们解决了这个问题： IP地址+协议+端口。

• 网络层的“IP地址”唯一标识了网络中的主机：这样就可以找到要将数据发送给哪台主机了。
• 传输层的“协议 + 端口”唯一标识主机中的应用程序：这样就可以找到要将数据发给那个应该程序了。

利用三元组（IP地址、协议、端口）就可以让计算机A确定将数据包发送给计算机B的应用程序了。

使用TCP/IP 协议的应用程序通常采用编程接口：UNIX BSD的套接字(socket)和UNIX System V的TLI(已经被淘汰)，来实现网络进程之间的通信。就目前而言， 几乎所有的应用程序都是采用的 Socket。

Socket

上面提到就目前而言，几乎所有的应用程序都是采用 Socket 来完成网络通信的。那什么是Socket呢？百度百科是这样定义的：

套接字（socket）是一个抽象层，应用程序可以通过它发送或接收数据，可对其进行像对文件一样的打开、读写和关闭等操作。套接字允许应用程序将I/O插入到网络中，并与网络中的其他应用程序进行通信。网络套接字是IP地址与端口的组合。

在TCP/IP四层模型中，我们并没有看到 Socket 影子，那它到底在哪里呢？ 又扮演什么角色呢？

Socket 并不是属于 TCP/IP 模型中的任何一层，它的存在只是为了让应用层能够更加简便地将数据传输给传输层，应用层不需要关注TCP/IP 协议的复杂内容。我们可以将其理解成一个接口，一个把复杂的TCP/IP协议族隐藏起来的接口，对于应用层而言，他们只需要简单地调用 Socket 接口就可以实现复杂的TCP/IP 协议，就像设计模式中的门面模式（ 将复杂的TCP\IP 协议族隐藏起来，对外提供统一的接口，是应用层能够更加容易地使用）。简单地说就是简单来说可以把 Socket理解成是应用层与TCP/IP协议族通信的抽象层、函数库。

下图是 Socket一次完整的通信流程图：

上图设计到的Socket 相关函数：

• socket()：返回套接字描述符
• connect()：建立连接
• bind()：一个本地协议地址赋予一个套接字
• linsten()：服务器监听端口连接
• accept()：应用程序接受完成3次握手的客户端连接
• send()、recv()、write()、read()：服务端与客户端互相发送数据
• colse()：关闭连接

探究SocketChannel

SocketChannel 是一个连接 TCP 网络Socket 的 Channel，我们可以认为它是对传统 Java Socket API的改进。它支持了非阻塞的读写。

SocketChannel具有如下特点

1. 对于已经存在的socket不能创建SocketChannel。
2. SocketChannel中提供的open接口创建的Channel并没有进行网络级联，需要使用connect接口连接到指定地址。
3. 未进行连接的SocketChannle执行I/O操作时，会抛出NotYetConnectedException。
4. SocketChannel支持两种I/O模式：阻塞式和非阻塞式。
5. SocketChannel支持异步关闭。如果SocketChannel在一个线程上read阻塞，另一个线程对该SocketChannel调用shutdownInput，则读阻塞的线程将返回-1表示没有读取任何数据；如果SocketChannel在一个线程上write阻塞，另一个线程对该SocketChannel调用shutdownWrite，则写阻塞的线程将抛出AsynchronousCloseException。

SocketChannel 的使用

1. 创建SocketChannel

要想使用 SocketChannel我们首先得创建它。创建SocketChannel的方式有两种：

// 方式 1
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("www.baidu.com", 80));

// 方式 2
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("www.baidu.com", 80));

2、连接校验

使用的SocketChannel必须是已连接的，如果使用一个未连接的SocketChannel，则会抛出 NotYetConnectedException。SocketChannel提供了四个方法来校验连接。

// 测试SocketChannel是否为open状态
socketChannel.isOpen();
// 测试SocketChannel是否已经被连接   
socketChannel.isConnected();
// 测试SocketChannel是否正在进行连接
socketChannel.isConnectionPending();
// 校验正在进行套接字连接的SocketChannel是否已经完成连接
socketChannel.finishConnect(); 

3、读操作

SocketChannel 提供了 read()方法用于读取数据：

public abstract int read(ByteBuffer dst) throws IOException;

public abstract long read(ByteBuffer[] dsts, int offset, int length) throws IOException;

public final long read(ByteBuffer[] dsts) throws IOException {
  return read(dsts, 0, dsts.length);
}

首先我们需要先分配一个 ByteBuffer，然后调用 read()方法，该方法会将数据从SocketChannel读入到 ByteBuffer中。

ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = socketChannel.read(buf);

read()方法会返回一个 int 值，该值表示读取了多少数据到 Buffer 中，如果返回 -1，则表示已经读到了流的末尾。

4、写操作

调用 SocketChannel的write()方法，可以向 SocketChannel 中写数据。

public abstract int write(ByteBuffer src) throws IOException;

public abstract long write(ByteBuffer[] srcs, int offset, int length) throws IOException;

public final long write(ByteBuffer[] srcs) throws IOException {
    return write(srcs, 0, srcs.length);
}

5、设置 I/O 模式

SocketChannel 支持阻塞和非阻塞两种 I/O 模式，调用 configureBlocking()方法即可：

socketChannel.configureBlocking(false);

false 表示非阻塞，true 表示阻塞。

6、关闭

当使用完 SocketChannel 后需要将其关闭，SocketChannel 提供了 close()来关闭 SocketChannel 。

socketChannel.close();

SocketChannel 源码分析

上面简单介绍了 SocketChannel 的使用，下面我们再来详细分析 SocketChannel 的源码。SocketChannel 实现 Channel 接口，它有一个核心子类 SocketChannel，该类实现了 SocketChannel 的大部分功能。如下（图有删减）

创建 SocketChannel

上面提到通过调用 open()方法就可以一个 SocketChannel 实例。

public static SocketChannel open() throws IOException {
    return SelectorProvider.provider().openSocketChannel();
}

我们看到它是通过 SelectorProvider 来创建 SocketChannel 的，provider() 方法会创建一个 SelectorProvider 实例，SelectorProvider 是 Selector 和 Channel 实例的提供者，它提供了创建 Selector、SocketChannel、ServerSocketChannel 实例的方法，采用 SPI 的方式实现。 SelectorProvider 我们在讲解 Selector 的时候在阐述。

provider 创建完成后调用 openSocketChannel() 来创建 SocketChannel。

public SocketChannel openSocketChannel() throws IOException {
    return new SocketChannelImpl(this);
}

从这了就可以看出 SocketChannelImpl 为 SocketChannel 的实现者。调用 SocketChannelImpl 的构造函数实例化一个 SocketChannel 对象。

SocketChannelImpl(SelectorProvider sp) throws IOException {
    super(sp);
    // 创建 Socket 并创建一个文件描述符与其关联
    this.fd = Net.socket(true);
    // 在注册 selector 的时候需要获取到文件描述符的值
    this.fdVal = IOUtil.fdVal(fd);
    // 设置状态为未连接
    this.state = ST_UNCONNECTED;
}

fd：文件夹描述符对象。

fdVal：fd 的 value。

文件描述符简称 fd，它是一个抽象概念，在 C 库编程中可以叫做文件流或文件流指针，在其它语言中也可以叫做文件句柄（handler），而且这些不同名词的隐含意义可能是不完全相同的。不过在系统层，我们统一把它叫做文件描述符。

state：状态，设置为未连接。它有如下 6 个值

private static final int ST_UNINITIALIZED = -1;
private static final int ST_UNCONNECTED = 0;
private static final int ST_PENDING = 1;
private static final int ST_CONNECTED = 2;
private static final int ST_KILLPENDING = 3;
private static final int ST_KILLED = 4;

连接服务器：connect()

调用 Connect() 方法可以链接远程服务器。

public boolean connect(SocketAddress sa) throws IOException {
    int localPort = 0;
    
    // 注意这里的加锁
    synchronized (readLock) {
        synchronized (writeLock) {
           // 确保当前 SocketChannel 是打开且未连接的
            ensureOpenAndUnconnected();
            InetSocketAddress isa = Net.checkAddress(sa);
            SecurityManager sm = System.getSecurityManager();
            if (sm != null)
                sm.checkConnect(isa.getAddress().getHostAddress(),
                                isa.getPort());
            // 这里的锁是注册和阻塞配置的锁
            synchronized (blockingLock()) {
                int n = 0;
                try {
                    try {
                        // 支持线程中断，通过设置当前线程的Interruptible blocker属性实现
                        begin();
                        // 
                        synchronized (stateLock) {
                           // 默认为 open, 除非调用了 close()
                            if (!isOpen()) {
                                return false;
                            }
                            // 只有未绑定本地地址也就是说未调用bind方法才执行
                            if (localAddress == null) {
                                NetHooks.beforeTcpConnect(fd,
                                                       isa.getAddress(),
                                                       isa.getPort());
                            }
                            // 记录当前线程
                            readerThread = NativeThread.current();
                        }
                        for (;;) {
                            InetAddress ia = isa.getAddress();
                            if (ia.isAnyLocalAddress())
                                ia = InetAddress.getLocalHost();
                            // 调用 Linux 的 connect 函数实现
                            // 如果采用堵塞模式，会一直等待，直到成功或出现异常
                            n = Net.connect(fd,
                                            ia,
                                            isa.getPort());
                            if (  (n == IOStatus.INTERRUPTED)
                                  && isOpen())
                                continue;
                            break;
                        }

                    } finally {
                        readerCleanup();
                        end((n > 0) || (n == IOStatus.UNAVAILABLE));
                        assert IOStatus.check(n);
                    }
                } catch (IOException x) {
                    // 出现异常，关闭 Channel
                    close();
                    throw x;
                }
                synchronized (stateLock) {
                    remoteAddress = isa;
                    if (n > 0) {
                        // n > 0,表示连接成功
                        // 连接成功，更新状态为ST_CONNECTED
                        state = ST_CONNECTED;
                        if (isOpen())
                            
                            localAddress = Net.localAddress(fd);
                        return true;
                    }
                    // 如果是非堵塞模式，而且未立即返回成功，更新状态为ST_PENDING
                    // 由此可见，该状态只有非堵塞时才会存在
                    if (!isBlocking())
                        state = ST_PENDING;
                    else
                        assert false;
                }
            }
            return false;
        }
    }
}

该方法的核心方法就在于 n = Net.connect(fd,ia,isa.getPort()); 该方法会一直调用到 native 方法去：

JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_connect0(JNIEnv *env, jclass clazz, jboolean preferIPv6,
                             jobject fdo, jobject iao, jint port)
{
    SOCKADDR sa;
    int sa_len = SOCKADDR_LEN;
    int rv;
    //地址转换为struct sockaddr格式
    if (NET_InetAddressToSockaddr(env, iao, port, (struct sockaddr *) &sa,
                                  &sa_len, preferIPv6) != 0)
    {
      return IOS_THROWN;
    }
   //传入 fd 和 sockaddr,与远程服务器建立连接，一般就是 TCP 三次握手
   //如果设置了 configureBlocking(false), 不会堵塞，否则会堵塞一直到超时或出现异常
    rv = connect(fdval(env, fdo), (struct sockaddr *)&sa, sa_len);
    if (rv != 0) { 
        // 0 表示连接成功，失败时通过 errno 获取具体原因
        if (errno == EINPROGRESS) {  //非堵塞，连接还未建立(-2)
            return IOS_UNAVAILABLE;
        } else if (errno == EINTR) {  //中断(-3)
            return IOS_INTERRUPTED;
        }
        return handleSocketError(env, errno); //出错
    }
    return 1; //连接建立,一般TCP连接连接都需要时间，因此除非是本地网络，一般情况下非堵塞模式返回IOS_UNAVAILABLE比较多；
}

读数据：read()

SocketChannel 提供 read() 方法读取数据。

public int read(ByteBuffer buf) throws IOException {
     synchronized (readLock) {
         // ...
         try {
             // ...
             for (;;) {
                 n = IOUtil.read(fd, buf, -1, nd);
                 if ((n == IOStatus.INTERRUPTED) && isOpen()) {
                     continue;
                 }
                 return IOStatus.normalize(n);
             }

         } finally {
             // ...
         }
     }
 }

核心方法就在于 IOUtil.read(fd, buf, -1, nd)。

static int read(FileDescriptor fd, ByteBuffer dst, long position,NativeDispatcher nd)
    throws IOException
{
    if (dst.isReadOnly())
        throw new IllegalArgumentException("Read-only buffer");
    if (dst instanceof DirectBuffer)
        // 使用直接缓冲区读取数据
        return readIntoNativeBuffer(fd, dst, position, nd);

    // 当不是使用直接内存时，则从线程本地缓冲获取一块临时的直接缓冲区存放待读取的数据
    ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
    try {
        int n = readIntoNativeBuffer(fd, bb, position, nd);
        bb.flip();
        if (n > 0)
            // 将直接缓冲区的数据写入到堆缓冲区中
            dst.put(bb);
        return n;
    } finally {
        // 使用完成后释放缓冲
        Util.offerFirstTemporaryDirectBuffer(bb);
    }
}

这里我们看到如果 ByteBuffer 是 DirectBuffer，则调用 readIntoNativeBuffer() 读取数据，如果不是则通过 getTemporaryDirectBuffer() 获取一个临时的直接缓冲区，然后调用 readIntoNativeBuffer()获取数据，然后将获取的数据写入 ByteBuffer 中。

private static int readIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb,long position, NativeDispatcher nd)
    throws IOException
{
    int pos = bb.position();
    int lim = bb.limit();
    assert (pos <= lim);
    int rem = (pos <= lim ? lim - pos : 0);

    if (rem == 0)
        return 0;
    int n = 0;
    if (position != -1) {
        n = nd.pread(fd, ((DirectBuffer)bb).address() + pos,rem, position);
    } else {
        n = nd.read(fd, ((DirectBuffer)bb).address() + pos, rem);
    }
    if (n > 0)
        bb.position(pos + n);
    return n;
}

写数据 write()方法和 read()方法大致一样，大明哥这里就不在阐述了，有兴趣的小伙伴自己去研究下。

ServerSocketChannel 与 SocketChannel 原理大同小异，这里就不展开讲述了，下篇文章我们开始研究第三个组件： Selector

参考资料

• https://zhuanlan.zhihu.com/p/180556309

---

Java 面试宝典是大明哥全力打造的 Java 精品面试题，它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题，而是一套完整的 Java 知识体系，一套你 Java 知识点的扫盲贴。

它的内容包括：

• 大厂真题：Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
• 原创内容：Java 面试宝典内容全部都是大明哥原创，内容全面且通俗易懂，回答部分可以直接作为面试回答内容。
• 持续更新：一次购买，永久有效。大明哥会持续更新 3+ 年，累计更新 1000+，宝典会不断迭代更新，保证最新、最全面。
• 覆盖全面：本宝典累计更新 1000+，从 Java 入门到 Java 架构的高频面试题，实现 360° 全覆盖。
• 不止面试：内容包含面试题解析、内容详解、知识扩展，它不仅仅只是一份面试题，更是一套完整的 Java 知识体系。
• 宝典详情：https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
• 宝典总览：https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
• 宝典进展：https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道，总字数 42w+。大明哥还在持续更新中，下图是大明哥在 2024-12 月份的更新情况：

想了解详情的小伙伴，扫描下面二维码加大明哥微信【daming091】咨询

同时，大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ，回复【面试题】 即可免费领取。