一、内核接收数据流程
1.网卡发现 MAC 地址符合，就将包收进来；发现 IP 地址符合，根据 IP 头中协议项，知道上一层是 TCP 协议；
2.DMA把TCP数据包copy到内核缓冲区；
3.触发CPU中断，中断程序摘除TCP头通过socket五要素(源IP/PORT、目的IP/PORT、协议)找到对应的socket文件，并把原始二进制数据报copy到socket接收缓冲区；
4.中断程序唤醒被阻塞的内核线程；
5.内核线程切换到用户线程把数据从socket接口缓冲区copy到应用内存；
二、中断处理流程
中断处理.png
I/O发出的信号的异常代码，拿到异常代码之后，CPU就会触发异常处理的流程。计算机在内存里会保存中断向量表，用来存放不同的异常代码对应的异常处理程序所在的地址。
CPU在拿到了异常码之后，会先把当前的程序执行的现场，保存到程序栈里面，然后根据异常码查询，找到对应的异常处理程序，最后把后续指令执行的指挥权，交给这个异常处理程序。
异常处理程序结束之后返回到原来指令执行的位置继续执行；
三、阻塞不占用 cpu
网卡何时接收到数据是依赖发送方和传输路径的，这个延迟通常都很高，是毫秒(ms)级别的。而应用程序处理数据是纳秒(ns)级别的。也就是说整个过程中，内核态等待数据，处理协议栈是个相对很慢的过程。这么长的时间里，用户态的进程是无事可做的，因此用到了“阻塞（挂起）”。
阻塞是进程调度的关键一环，指的是进程在等待某事件发生之前的等待状态。请看下表，在 Linux 中，进程状态大致有 7 种（在
include/linux/sched.h 中有更多状态）：
从说明中可以发现，“可运行状态”会占用 CPU 资源，另外创建和销毁进程也需要占用 CPU 资源（内核）。重点是，当进程被"阻塞/挂起"时，是不会占用
CPU 资源的。
为了支持多任务，Linux 实现了进程调度的功能（CPU
时间片的调度）。而这个时间片的切换，只会在“可运行状态”的进程间进行。因此“阻塞/挂起”的进程是不占用 CPU 资源的。
四、工作队列和等待队列
工作队列和等待队列.png
工作队列：为了方便时间片的调度，所有“可运行状态”状态的进程组成的队列；
fd文件列表：内核打开的文件句柄，Linux一切皆文件，用户线程执行创建Socket时内核就会创建一个由文件系统管理的sock对象；
sock：socket内核中的数据结构，主要包含发送缓冲区、接收缓冲区、等待队列；
struct sock { __u32 daddr; /* 外部IP地址 */ __u32 rcv_saddr; /* 绑定的本地IP地址 */ __u16 dport; /* 目标端口 */ __u16 sport; /* 源端口 */ unsigned short family; /* 地址簇 */ int rcvbuf; /* 接受缓冲区长度(单位：字节) */ struct sk_buff_head receive_queue; /* 接收包队列 */ int sndbuf; /* 发送缓冲区长度(单位：字节) */ struct sk_buff_head write_queue; /* 包发送队列 */ wait_queue_head_t *sleep; /* 等待队列，通常指向socket的wait域 */ ......}等待队列：等待当前socket的线程；
工作队列中线程执行到阻塞操作等待socket时，会从工作队列中移除，移动到该socket的等待队列中；当socket接收到数据后，操作系统将该socket等待队列上的进程重新放回到工作队列，该进程变成运行状态，继续执行代码。
五、BIO
代码语言：txt复制public static void main(String[] args) throws IOException { ServerSocket serverSocket = new ServerSocket(9000); while (true) { // 没有连接-阻塞 Socket socket = serverSocket.accept(); byte[] bytes = new byte[1024]; InputStream inputStream = socket.getInputStream(); while (true) { // 没有数据-阻塞 int read = inputStream.read(bytes); if (read != -1) { System.out.println(new String(bytes, 0, read)); } else { break; } } socket.close(); }}BIO模式存在两个阻塞点，一个时accept阻塞等待客户端连接，一个是阻塞等待socket请求数据；简单跟以下源码就会发现
new ServerSocket(9000)最终通过
int newfd = socket0(stream, false /*v6 Only*/);调用Linuxint socket(int domain, int type, int protocol);创建服务端socket；bind0(nativefd, address, port, exclusiveBind);调用Linuxint bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);绑定端口；listen0(nativefd, backlog);调用Linuxint listen(int sockfd, int backlog);设置监听；
接下来serverSocket.accept()最终通过
newfd = accept0(nativefd, isaa);调用Linuxint accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);阻塞等待客户端连接，会创建一个socket用于跟该客户端进行通信，返回socket文件描述符fd；
最后inputStream.read(bytes);调用Linux `ssize_t recv(int sockfd, void *buf,
size_t len, int flags);`阻塞从socket读缓冲区读取客户端请求数据；
可以通过 man 命令查看Linux 系统调用方法具体描述；
通过传统BIO的操作方式可以看出一个请求必须要创建一个内核线程进行处理，recv只能监视单个socket，当并发比较高时就会消耗大量系统资源，也就是所谓的C10K问题；那么如何解决这个问题呢？后面出现的多路复用
select / poll / epoll思路都是使用一个线程来处理若干个连接(监视若干个socket)，类似餐厅服务员的角色。
六、select
select 方案是用一个 fd_set 结构体来告诉内核同时监控多个socket，当其中有socket的状态发生变化或超时，则调用返回。之后应用使用
FD_ISSET 来逐个查看是哪个socket的状态发生了变化。
6.1 select原理
以下面select伪代码为例：
代码语言：txt复制int s = socket(AF_INET, SOCK_STREAM, 0); bind(s, ...)listen(s, ...)//接受客户端连接int c = accept(s, ...)int readSet[] = 存放需要监听的socket文件描述符while(1){ int n = select(..., readSet, ...) for(int i=0; i < readSet.count; i ){ if(FD_ISSET(readSet[i], ...)){ //fds[i]的数据处理 } }}select 系统调用函数# maxfd：表示的是待测试的描述符基数，它的值是待测试的最大描述符加 1# readset：读描述符集合# writeset：写描述符集合# exceptset：错误描述符集合# timeout：超时时间int select(int maxfd, fd_set *readset, fd_set *writeset, fd_set *exceptset, const struct timeval *timeout);上面示例代码中，先准备一个数组 readSet 存放着所有需要监视读事件的socket。然后调用select，如果 readSet
中的所有socket都没有数据，select会阻塞，直到有一个socket接收到数据，select返回，唤醒线程。用户可以遍历
readSet，通过FD_ISSET (对这个数组进行检测，判断出对应socket的元素 readSetfd是 0 还是 1)判断具体哪个 socket
收到数据，然后做出处理。
6.2 select流程
select-1.png
线程 A 调用 select readSet 数组元素为sock1、sock2、sock3 ，此时3个socket 都没有数据可读，就把线程A
从工作队列中移除，并分别添加到3个sock 的等待队列中；
select-2.png
当3个sock中任意一个有数据可读时，中断程序都会把线程A
从所有sock等待队列中移除并重新加入工作队列，等待cpu时间片继续执行(即从select中返回)。但是此时线程A
并不知道哪个socket有数据，于是还要遍历readSet使用 FD_ISSET 来逐个查看是哪个socket的有数据可读。
6.3 select的缺点
1、每次调用select都需要将线程加入到所有监视socket的等待队列，每次唤醒都需要从每个队列中移除。这里涉及了两次遍历，而且每次都要将整个fd_set列表传递给内核，有一定的开销。正是因为遍历操作开销大，出于效率的考量，才会规定select的最大监视数量，默认只能监视1024个socket。
2、由于只有一个字段记录关注和发生事件，每次调用之前要重新初始化 fd_set 结构体。
3、线程被唤醒后，程序并不知道哪些socket收到数据，还需要遍历一次。
七、poll
总结以下 select 的缺点就是句柄上限 重复初始化 逐个排查所有文件句柄状态效率不高。poll 主要解决 select
的前两个问题：通过改变跟内核交互的数据结构突破了文件描述符的限制，同时使用不同字段分别标注关注事件和发生事件，来避免重复初始化。
代码语言：txt复制int poll(struct pollfd *fds, unsigned long nfds, int timeout); # fd：描述符 fd# events：描述符上待检测的事件类型events，注意这里的 events 可以表示多个不同的事件，# 具体的实现可以通过使用二进制掩码位操作来完成，例如，POLLIN 和 POLLOUT 可以表示读和写事件# revents：struct pollfd { int fd; /* file descriptor */ short events; /* events to look for */ short revents; /* events returned */ };和 select 非常不同的地方在于：
- poll 每次检测之后的结果不会修改原来的传入值，而是将结果保留在 revents 字段中，这样就不需要每次检测完都得重置待检测的描述字和感兴趣的事件。
- 在 select 里面，文件描述符的个数已经随着 fd_set 的实现而固定，没有办法对此进行配置；而在 poll 函数里，可以控制 pollfd 结构的数组大小，这意味着可以突破原来 select 函数最大描述符的限制，在这种情况下，应用程序调用者需要分配 pollfd 数组并通知 poll 函数该数组的大小。
八、epoll
epoll 使用 eventpoll 作为中间层，线程不用加入在被监视的每一个 socket 阻塞队列中，也不用再遍历 socket
列表查看哪些有事件发生。
epoll提供了三个函数：
代码语言：txt复制# 创建了一个 epoll 实例 epolleventint epoll_create(int size);# 往这个 epoll 实例增加或删除监控的事件# epfd：epoll_create 创建的 epoll 实例句柄# op：增加还是删除一个监控事件# fd：注册的事件的文件描述符# event：注册的事件类型，并且可以在这个结构体里设置用户需要的数据int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);# 类似之前的 poll 和 select 函数，调用者进程被挂起，在等待内核 I/O 事件的分发# epfd：实例描述字，也就是 epoll 句柄# events：用户空间需要处理的 I/O 事件，这是一个数组，数组的大小由 epoll_wait 的返回值决定，这个数组的每个元素都是一个需要待处理的 I/O 事件，其中 events 表示具体的事件类型，事件类型取值和 epoll_ctl 可设置的值一样，这个 epoll_event 结构体里的 data 值就是在 epoll_ctl 那里设置的 data，也就是用户空间和内核空间调用时需要的数据# maxevents：大于 0 的整数，表示 epoll_wait 可以返回的最大事件值# timeout：阻塞调用的超时值，如果这个值设置为 -1，表示不超时；如果设置为 0 则立即返回，即使没有任何 I/O 事件发生int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);8.1 代码示例
代码语言：txt复制public class ServerConnect { private static final int BUF_SIZE = 1024; private static final int PORT = 8080; private static final int TIMEOUT = 3000; public static void main(String[] args) { selector(); } public static void selector() { Selector selector = null; ServerSocketChannel ssc = null; try { // 打开一个Channel ssc = ServerSocketChannel.open(); // 将Channel绑定端口 ssc.socket().bind(new InetSocketAddress(PORT)); // 设置Channel为非阻塞，如果设置为阻塞，其实和BIO差不多了。 ssc.configureBlocking(false); // 打开一个Slectore selector = Selector.open(); // 向selector中注册Channel和感兴趣的事件 ssc.register(selector, SelectionKey.OP_ACCEPT); while (true) { // selector监听事件，select会被阻塞，直到selector监听的channel中有事件发生或者超时，会返回一个事件数量 //TIMEOUT就是超时时间，selector初始化的时候会添加一个用于主动唤醒的pipe，待会源码分析会说 if (selector.select(TIMEOUT) == 0) { System.out.println("=="); continue; } /** * SelectionKey的组成是selector和Channel
* 有事件发生的channel会被包装成selectionKey添加到selector的publicSelectedKeys属性中
* publicSelectedKeys是SelectionKey的Set集合
*下面这一部分遍历，就是遍历有事件的channel
*/
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
if (key.isAcceptable()) {
handleAccept(key);
}
if (key.isReadable()) {
handleRead(key);
}
if (key.isWritable() && key.isValid()) {
handleWrite(key);
}
if (key.isConnectable()) {
System.out.println("isConnectable = true");
}
//每次使用完，必须将该SelectionKey移除，否则会一直存储在publicSelectedKeys中
//下一次遍历又会重复处理
iter.remove();
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (selector != null) {
selector.close();
}
if (ssc != null) {
ssc.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void handleAccept(SelectionKey key) throws IOException {
ServerSocketChannel ssChannel = (ServerSocketChannel) key.channel();
SocketChannel sc = ssChannel.accept();
sc.configureBlocking(false);
sc.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(BUF_SIZE));
}
public static void handleRead(SelectionKey key) throws IOException {
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer buf = (ByteBuffer) key.attachment();
long bytesRead = sc.read(buf);
while (bytesRead > 0) {
buf.flip();
while (buf.hasRemaining()) {
System.out.print((char) buf.get());
}
System.out.println();
buf.clear();
bytesRead = sc.read(buf);
}
if (bytesRead == -1) {
sc.close();
}
}
public static void handleWrite(SelectionKey key) throws IOException {
ByteBuffer buf = (ByteBuffer) key.attachment();
buf.flip();
SocketChannel sc = (SocketChannel) key.channel();
while (buf.hasRemaining()) {
sc.write(buf);
}
buf.compact();
}
}调用到内核的流程如下：
1、ssc = ServerSocketChannel.open() -> EPollSelectorProvider -> int
socket(int domain, int type, int protocol);
2、ssc.socket().bind(newInetSocketAddress(PORT)); -> int bind(int sockfd,
const struct sockaddr _addr, socklen_t addrlen);
3、getImpl().listen(backlog); -> int listen(int sockfd, int backlog);
4、ssc.configureBlocking(false); -> int fcntl(int fd, int cmd, ... /_ arg
*/ );
5、selector=Selector.open(); -> EPollSelectorImpl epollCreate() -> int
epoll_create(int size);
6、ssc.register(selector,SelectionKey.OP_ACCEPT); -> int epoll_ctl(int
epfd, int op, int fd, struct epoll_event *event);
while
7、selector.select(TIMEOUT) -> int epoll_wait(int epfd, struct epoll_event
*events, int maxevents, int timeout);
8.2 epoll流程
epoll.png
如图，eventpoll 主要包含3个结构
监视队列：epoll_ctl添加或删除所要监听的socket。例如：图中通过epoll_ctl添加sock1、sock2的监视，内核会将eventpoll添加到这socket的等待队列中(图中虚线)。为了方便的添加和移除，还要便于搜索，以避免重复添加，使用红黑树结构进行存储(搜索、插入和删除时间复杂度都是O(log(N))，效率较好。等待队列：eventpoll也是一个文件句柄，因此也有等待队列，记录阻塞的线程，例如：图中线程A 执行了 epoll_wait 操作，被从工作队列中移除，然后被eventpoll 的等待队列引用；就绪队列：当某个socket有事件发生时，中断处理程序就会把该socket加入到就绪队列，同时唤醒eventpoll 阻塞队列中的线程，此时线程只需要遍历就绪队列就可以知道哪个socket有事件发生，例如：图中sock2有数据到来时，中断处理程序先把sock2 放入就绪队列中，然后唤醒等待队列中的线程A，这时线程A 被重新加入工作队列中，等到CPU时间片轮询到线程A时，遍历就绪队列中的socket进行处理。
