socket-io的底层实现设计原理

亦山

发布于 2021-04-22 15:35:07

8000

发布于 2021-04-22 15:35:07

文章被收录于专栏：Spring Cloud设计原理Spring Cloud设计原理

前言

上一篇文章《漫谈socket-io的基本原理》用了现实非常浅显的例子，尽可能地阐释非阻塞、阻塞、多线程、多路复用poll和 epoll 背后演进的整体思考脉络，将有助于读者从宏观的角度把握住socket-io的本质。本文将聚焦在JDK socket-io 的多路复用 poll/epoll 的实现原理，可能比较枯燥复杂，为了降低理解成本，作者尽可能循序渐进，控制每个步骤的信息量。

如果文章不错，欢迎分享转载，关注公众号：亦山札记(louluan_note)

现实生活中的例子

上一篇文章《漫谈socket-io的基本原理》中提到的餐厅中服务员Amy 的工作模式，实际上和真正的Socket 工作模式非常的相似：

餐厅	Socket
服务员Amy 前台接待，如果没有等到顾客，就一直阻塞；	ServerSocket 在监听服务端口，等待Socket 连接，如果没有连接，则阻塞等待；
服务员Amy 等待顾客点餐，如果顾客没点好，就一直阻塞等待	获取socket.inputStream() 输入流，如果没有输入，则阻塞等待
服务员Amy 给顾客上菜，如果餐桌已满放不下，则阻塞等待	往socket.outputStream() 输出流中写数据，如果输出流满，则阻塞等待
前台和餐桌安排闹铃，条件满足后通知Amy,但是Amy 并不知道具体是谁发起的，需要依次去前台和各个餐桌上确认的过程	socket的多路复用 poll的工作模式
前台和餐桌安排闹铃，条件满足后通知Amy,但是Amy 知道具体是谁发起的,直接到发起前台或者餐桌服务的过程	socket的多路复用 epoll的工作模式

对应地，ServerSocket 端的socket工作模式大概如下图所示：

典型的服务端Socket工作流程是：

监听指定端口，等待连接这个过程可能会一直阻塞；
接收到客户端连接后，创建Socket对象，指定或者随机一个端口号，以表示和 remote socket 的连接；
socket 尝试获取输入流InputStream,如果没有远程socket没有数据，则一直阻塞
socket 尝试往输出流OutputStream 输出数据，如果输出流已满，则一直阻塞；

接下来将介绍在多路复用模型下的socket 工作模式。

多路复用选择器-Selector的原理

很多人在讲多路复用实现时，倾向把操作系统的一些底层如Linux的poll 和epoll 一起拿来讲，整体感觉边界不是很清晰，理解成本比较高。为了界定清楚，作者将socket工作过程做了系统边界区分，即:Java编程区、操作系统内核区、网络区，整体的工作模式如下所示：

先看系统边界：

操作系统内核区和网络 无论是Windows还是Linux 系统，底层和网络socket 通信，都会通过句柄(File Descriptor, 也可以叫做文件描述符)来操作；Java编程区 创建的每一个socket对象，操作系统会分配一个FD , 后续的IO操作，都是通过Java本地方法调用传入 FD 来操作 socket。
Java 编程区 Java编程区 主要是对多路复用选择器的抽象，Channel 的注册管理；当多路复用选择器做选择操作时，具体能够选中哪些socket的什么操作，底层是Java 本地方法调用，具体操作系统是通过poll 还是epoll的方式，JDK是决定不了的，也不要关心。

Selector 的组成结构

Selector 内部维护了一个PollArrayWrapper 的连续内存数组，用来动态维护socket 的注册关系以及socket的IO 操作 ready情况:

FD句柄(File Descriptor)，int类型(4字节)，socketChannel 注册时对应socket 的FD；
events，short类型(2字节)，socketChannel注册时对应的interestOps 经过转换后存储到events中，JDK中定义了selector 可以注册的操作类型(OPS)如下所示：

操作	名称	位值
OP_READ	数据读	0000 0001
OP_WRITE	数据写	0000 0100
OP_CONNECT	Socket连接(针对客户端socket)	0000 1000
OP_ACCEPT	Socket 接受连接(针对客户端 socket)	0001 0000

而每个操作系统如windows、linux 的JDK内部实现对events的位定义会有所区别，比如笔者的windows，定义的如下几种events:

操作	名称	位值(不同计算机可能有差异)
POLLIN	普通或优先级带数据可读	768
POLLOUT	普通数据可写	16
POLLERR	发生错误	1
POLLHUP	发生挂起	2
POLLNVAL	描述字不是一个打开的文件	4
POLLCONN	连接就绪	8192

revents，short类型(2字节)，当调用 selector.select() 时，会触发本地方法调用获取注册的socket的操作就绪情况，会更新到revents 中。调用Set<SelectionKey> selectedKeys()，就是根据 events(注册的操作) 和revents(就绪操作) 通过一定的规则按位取 & 来判断是否匹配被选中的。注意revents的值和events的值并不完全一样，revents 记录的时底层网络请求的操作。

Selector 的工作流程

多路复用选择器(Selector) 的工作流程整体可以分为三步：

第一步：Channel注册到 Selector 上；如果是ServerSocketChannel 则注册 SelectionKey.OP_ACCEPT 操作到Selector，如果是SocketChannel 则可注册SelectionKey.OP_CONNECT、SelectionKey.OP_READ、SelectionKey.OP_WRITE到Selector上；

Selector 内部维护了一个PollArrayWrapper的连续数组，会将对应SocketChannel的FD 写入到 FD区域，将注册的操作Ops 经过内部按位转换成 16位数值，存在events中：

以 windows的JDK实现为例，SocketChannel往Selector注册时，转换events 代码实现如下所示：

    public void translateAndSetInterestOps(int var1, SelectionKeyImpl var2) {
        int var3 = 0;
        if ((var1 & 1) != 0) {
            var3 |= Net.POLLIN;
        }

        if ((var1 & 4) != 0) {
            var3 |= Net.POLLOUT;
        }

        if ((var1 & 8) != 0) {
            var3 |= Net.POLLCONN;
        }

        var2.selector.putEventOps(var2, var3);
    }

第二步：Selector.select()，选择发生的操作Ready事件；如果没有触发操作Ready事件，则一直阻塞。如果Ready事件发生，则select() 底层会把各个FD背后的channel Ready 情况写入到PollArrayWrapper对应的revents中。

select() 方法底层对于不同的JDK实现，采用的策略可能会有所不同。对于windows和 linux 2.6之前的版本，使用的时poll模式；而对于linux 2.6 及以后的版本，则使用的是epoll模式。poll 和epoll 简单来讲最大的区别在于poll 会把所有的句柄全部遍历一遍来看有没有发生操作ready事件, 而epoll 只会遍历发生了操作ready事件的句柄，对于大量socket连接处理的场景性能会更高。

备注：本文的重点不是解释 poll和epoll 的底层实现原理，因为这个是纯粹的不同的操作系统内核的实现，有兴趣的同学可以看下知乎的这边文章《如果这篇文章说不清epoll的本质，那就过来掐死我吧！》

第三步：获取被选择的Key: selector.selectedKeys(). 调用了此方法，会把 PollArrayWrapper 内表示的所有句柄的events 和 revents 进行匹配，看下感兴趣的事件(在events中) 有没有Ready(在revents)中，通过一定的按位 & 计算，最终转换成SelectionKey的OPS(OP_ACCEPT、OP_CONNECT、OP_READ、OP_WRITE)。

以windows为例，当执行了selector.select 之后，根据revents 的值计算readyOps的过程：

public boolean translateReadyOps(int var1, int var2, SelectionKeyImpl var3) {
        int var4 = var3.nioInterestOps();//感兴趣的OPS
        int var5 = var3.nioReadyOps();
        int var6 = var2;
        if ((var1 & Net.POLLNVAL) != 0) {
            return false;
        } else if ((var1 & (Net.POLLERR | Net.POLLHUP)) != 0) {
            var3.nioReadyOps(var4);
            this.readyToConnect = true;
            return (var4 & ~var5) != 0;
        } else {
            if ((var1 & Net.POLLIN) != 0 && (var4 & 1) != 0 && this.state == 2) {
                var6 = var2 | 1;
            }

            if ((var1 & Net.POLLCONN) != 0 && (var4 & 8) != 0 && (this.state == 0 || this.state == 1)) {
                var6 |= 8;
                this.readyToConnect = true;
            }

            if ((var1 & Net.POLLOUT) != 0 && (var4 & 4) != 0 && this.state == 2) {
                var6 |= 4;
            }

            var3.nioReadyOps(var6);
            return (var6 & ~var5) != 0;
        }
    }

实战案例

package org.luanlouis.socket.nio;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;

public class Main {

    public static void main(String[] args) throws Exception{

        DefaultSocketHandler defaultSocketHandler = new DefaultSocketHandler();
        ServerSocketChannel serverSocketChannel  = ServerSocketChannel.open();
        //设为noblocking
        serverSocketChannel.configureBlocking(false);

        SocketAddress socketAddress = new InetSocketAddress(8080);
        serverSocketChannel.bind(socketAddress);

        Selector selector = Selector.open();

        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        // 尝试多路复用选择器选择,如果没有Ready时间发生，则一直阻塞
        while (selector.select()>0){

            Set<SelectionKey> selectionKeySet = selector.selectedKeys();

            for (SelectionKey selectionKey : selectionKeySet) {
                // server socket 准备好接受连接，获取连接
                if(selectionKey.isAcceptable()){
                    SocketChannel socketChannel =((ServerSocketChannel)selectionKey.channel()).accept();
                    if(null != socketChannel){
                        //监听读写
                        socketChannel.register(selector,SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                    }
                }

                // socket 数据可读
                if(selectionKey.isReadable()){
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(5000);
                    socketChannel.read(buffer);
                    defaultSocketHandler.onReceiveData(buffer,socketChannel);
                }

                // socket 数据可写
                if(selectionKey.isWritable()){
                    ByteBuffer buffer = ByteBuffer.allocate(5000);
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    //处理数据写入请求
                    defaultSocketHandler.onReadySendData(buffer,socketChannel);
                }
            }
        }
    }
}