python_异步io

小鱼儿 2022-01-29 13:13 214阅读 0赞

# #

## 协程 ##

协程，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：**协程是一种用户态的轻量级线程**。

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：

协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。

协程的好处：

*  无需线程上下文切换的开销
 *  无需原子操作锁定及同步的开销
    
     *  　　"原子操作(atomic operation)是不需要synchronized"，所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序是不可以被打乱，或者切割掉只执行部分。视作整体是原子性的核心。
 *  方便切换控制流，简化编程模型
 *  高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点：

*  无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
 *  进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

使用yield实现协程操作例子

import time
    import queue
    def consumer(name):
        print("--->starting eating baozi...")
        while True:
            new_baozi = yield
            print("[%s] is eating baozi %s" % (name,new_baozi))
            #time.sleep(1)
     
    def producer():
     
        r = con.__next__()
        r = con2.__next__()
        n = 0
        while n < 5:
            n +=1
            con.send(n)
            con2.send(n)
            print("\033[32;1m[producer]\033[0m is making baozi %s" %n )
     
     
    if __name__ == '__main__':
        con = consumer("c1")
        con2 = consumer("c2")
        p = producer()

**协程**

1.  **必须在只有一个单线程里实现并发**
2.  **修改共享数据不需加锁**
3.  **用户程序里自己保存多个控制流的上下文栈**
4.  **一个协程遇到IO操作自动切换到其它协程**

### **Greenlet** ###

greenlet是一个用C实现的协程模块，相比与python自带的yield，它可以使你在任意函数之间随意切换，而不需把这个函数先声明为generator

from greenlet import greenlet
     
     
    def test1():
        print(12)
        gr2.switch()
        print(34)
        gr2.switch()
     
     
    def test2():
        print(56)
        gr1.switch()
        print(78)
     
     
    gr1 = greenlet(test1)
    gr2 = greenlet(test2)
    gr1.switch()

感觉确实用着比generator还简单了呢，但好像还没有解决一个问题，就是遇到IO操作，自动切换，对不对？

### Gevent  ###

Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是**Greenlet**, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

import gevent
     
    def func1():
        print('\033[31;1m李闯在跟海涛搞...\033[0m')
        gevent.sleep(2)
        print('\033[31;1m李闯又回去跟继续跟海涛搞...\033[0m')
     
    def func2():
        print('\033[32;1m李闯切换到了跟海龙搞...\033[0m')
        gevent.sleep(1)
        print('\033[32;1m李闯搞完了海涛，回来继续跟海龙搞...\033[0m')
     
     
    gevent.joinall([
        gevent.spawn(func1),
        gevent.spawn(func2),
    ])

输出：

李闯在跟海涛搞...  
李闯切换到了跟海龙搞...  
李闯搞完了海涛，回来继续跟海龙搞...  
李闯又回去跟继续跟海涛搞...

**遇到IO阻塞时会自动切换任务**

from gevent import monkey; monkey.patch_all()
    import gevent
    from  urllib.request import urlopen
     
    def f(url):
        print('GET: %s' % url)
        resp = urlopen(url)
        data = resp.read()
        print('%d bytes received from %s.' % (len(data), url))
     
    gevent.joinall([
            gevent.spawn(f, 'https://www.python.org/'),
            gevent.spawn(f, 'https://www.yahoo.com/'),
            gevent.spawn(f, 'https://github.com/'),
    ])

**通过gevent实现单线程下的多socket并发**

**server side **

import sys
    import socket
    import time
    import gevent
     
    from gevent import socket,monkey
    monkey.patch_all()
     
     
    def server(port):
        s = socket.socket()
        s.bind(('0.0.0.0', port))
        s.listen(500)
        while True:
            cli, addr = s.accept()
            gevent.spawn(handle_request, cli)
     
     
     
    def handle_request(conn):
        try:
            while True:
                data = conn.recv(1024)
                print("recv:", data)
                conn.send(data)
                if not data:
                    conn.shutdown(socket.SHUT_WR)
     
        except Exception as  ex:
            print(ex)
        finally:
            conn.close()
    if __name__ == '__main__':
        server(8001)

**client side 　　**

import socket
     
    HOST = 'localhost'    # The remote host
    PORT = 8001           # The same port as used by the server
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((HOST, PORT))
    while True:
        msg = bytes(input(">>:"),encoding="utf8")
        s.sendall(msg)
        data = s.recv(1024)
        #print(data)
     
        print('Received', repr(data))
    s.close()

--------------------

### 论事件驱动与异步IO ###

通常，我们写服务器处理模型的程序时，有以下几种模型：

（1）每收到一个请求，创建一个新的进程，来处理该请求；

（2）每收到一个请求，创建一个新的线程，来处理该请求；

（3）每收到一个请求，放入一个事件列表，让主进程通过非阻塞I/O方式来处理请求

上面的几种方式，各有千秋，

第（1）中方法，由于创建新的进程的开销比较大，所以，会导致服务器性能比较差,但实现比较简单。

第（2）种方式，由于要涉及到线程的同步，有可能会面临[死锁][Link 1]等问题。

第（3）种方式，在写应用程序代码时，逻辑比前面两种都复杂。

综合考虑各方面因素，一般普遍认为第（3）种方式是大多数[网络服务器][Link 2]采用的方式

在UI编程中，常常要对鼠标点击进行相应，首先如何获得鼠标点击呢？  
**方式一：创建一个线程，该线程一直循环检测是否有鼠标点击，那么这个方式有以下几个缺点**：  
1. CPU资源浪费，可能鼠标点击的频率非常小，但是扫描线程还是会一直循环检测，这会造成很多的CPU资源浪费；如果扫描鼠标点击的接口是阻塞的呢？  
2. 如果是堵塞的，又会出现下面这样的问题，如果我们不但要扫描鼠标点击，还要扫描键盘是否按下，由于扫描鼠标时被堵塞了，那么可能永远不会去扫描键盘；  
3. 如果一个循环需要扫描的设备非常多，这又会引来响应时间的问题；  
所以，该方式是非常不好的。

**方式二：就是事件驱动模型**  
目前大部分的UI编程都是事件驱动模型，如很多UI平台都会提供onClick()事件，这个事件就代表鼠标按下事件。事件驱动模型大体思路如下：  
1. 有一个事件（消息）队列；  
2. 鼠标按下时，往这个队列中增加一个点击事件（消息）；  
3. 有个循环，不断从队列取出事件，根据不同的事件，调用不同的函数，如onClick()、onKeyDown()等；  
4. 事件（消息）一般都各自保存各自的处理函数指针，这样，每个消息都有独立的处理函数；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70][]

事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是（单线程）同步以及多线程编程。

让我们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展示了随着时间的推移，这三种模式下程序所做的工作。这个程序有3个任务需要完成，每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1][]

在单线程同步模型中，任务按照顺序执行。如果某个任务因为I/O而阻塞，其他所有的任务都必须等待，直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系，但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。

在多线程版本中，这3个任务分别在独立的线程中执行。这些线程由操作系统来管理，在多处理器系统上可以并行处理，或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。与完成类似功能的同步程序相比，这种方式更有效率，但程序员必须写代码来保护共享资源，防止其被多个线程同时访问。多线程程序更加难以推断，因为这类程序不得不通过线程同步机制如锁、可重入函数、线程局部存储或者其他机制来处理线程安全问题，如果实现不当就会导致出现微妙且令人痛不欲生的bug。

在事件驱动版本的程序中，3个任务交错执行，但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时，注册一个回调到事件循环中，然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件，当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽可能的得以执行而不需要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为，因为程序员不需要关心线程安全问题。

当我们面对如下的环境时，事件驱动模型通常是一个好的选择：

1.  程序中有许多任务，而且…
2.  任务之间高度独立（因此它们不需要互相通信，或者等待彼此）而且…
3.  在等待事件到来时，某些任务会阻塞。

当应用程序需要在任务间共享可变的数据时，这也是一个不错的选择，因为这里不需要采用同步处理。

网络应用程序通常都有上述这些特点，这使得它们能够很好的契合事件驱动编程模型。

**事件模型的实现**

## **Select\\Poll\\Epoll异步IO　** ##

同步IO和异步IO，阻塞IO和非阻塞IO分别是什么，到底有什么区别？不同的人在不同的上下文下给出的答案是不同的。所以先限定一下本文的上下文。

本文讨论的背景是Linux环境下的network IO。

### 概念说明 ###

在进行解释之前，首先要说明几个概念：  
\- 用户空间和内核空间  
\- 进程切换  
\- 进程的阻塞  
\- 文件描述符  
\- 缓存 I/O

**用户空间**与**内核空间**

现在操作系统都是采用虚拟存储器，那么对32位操作系统而言，它的寻址空间（虚拟存储空间）为4G（2的32次方）。操作系统的核心是内核，独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的所有权限。为了保证用户进程不能直接操作内核（kernel），保证内核的安全，操心系统将虚拟空间划分为两部分，一部分为内核空间，一部分为用户空间。针对linux操作系统而言，将最高的1G字节（从虚拟地址0xC0000000到0xFFFFFFFF），供内核使用，称为内核空间，而将较低的3G字节（从虚拟地址0x00000000到0xBFFFFFFF），供各个进程使用，称为用户空间。

**进程切换**

为了控制进程的执行，内核必须有能力挂起正在CPU上运行的进程，并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。因此可以说，任何进程都是在操作系统内核的支持下运行的，是与内核紧密相关的。

从一个进程的运行转到另一个进程上运行，这个过程中经过下面这些变化：  
1. 保存处理机上下文，包括程序计数器和其他寄存器。  
2. 更新PCB信息。

3. 把进程的PCB移入相应的队列，如就绪、在某事件阻塞等队列。  
4. 选择另一个进程执行，并更新其PCB。  
5. 更新内存管理的数据结构。  
6. 恢复处理机上下文。

需要了解更多,可以参考[进程切换][Link 3]

**进程的阻塞**

正在执行的进程，由于期待的某些事件未发生，如请求系统资源失败、等待某种操作的完成、新数据尚未到达或无新工作做等，则由系统自动执行阻塞原语(Block)，使自己由运行状态变为阻塞状态。可见，进程的阻塞是进程自身的一种主动行为，也因此只有处于运行态的进程（获得CPU），才可能将其转为阻塞状态。`当进程进入阻塞状态，是不占用CPU资源的`。

**文件描述符fd**

文件描述符（File descriptor）是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

**缓存IO**

缓存 I/O 又被称作标准 I/O，大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中，操作系统会将 I/O 的数据缓存在文件系统的页缓存（ page cache ）中，也就是说，数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

缓存 I/O 的缺点：  
数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作，这些数据拷贝操作所带来的 CPU 以及内存开销是非常大的。

### IO模式 ###

刚才说了，对于一次IO访问（以read举例），数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说，当一个read操作发生时，它会经历两个阶段：  
1. 等待数据准备 (Waiting for the data to be ready)  
2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正式因为这两个阶段，linux系统产生了下面五种网络模式的方案。  
\- 阻塞 I/O（blocking IO）  
\- 非阻塞 I/O（nonblocking IO）  
\- I/O 多路复用（ IO multiplexing）  
\- 信号驱动 I/O（ signal driven IO）  
\- 异步 I/O（asynchronous IO）

注：由于signal driven IO在实际中并不常用，所以我这只提及剩下的四种IO Model。

**阻塞 I/O（blocking IO）**

在linux中，默认情况下所有的socket都是blocking，一个典型的读操作流程大概是这样：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2][]

当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据（对于网络IO来说，很多时候数据在一开始还没有到达。比如，还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来）。这个过程需要等待，也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边，整个进程会被阻塞（当然，是进程自己选择的阻塞）。当kernel一直等到数据准备好了，它就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才解除block的状态，重新运行起来。

\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_blocking IO的特点就是在IO执行的两个阶段都被block了。

**非阻塞 I/O（nonblocking IO）**

linux下，可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时，流程是这个样子：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3][]

当用户进程发出read操作时，如果kernel中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个error。从用户进程角度讲 ，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，它就知道数据还没有准备好，于是它可以再次发送read操作。一旦kernel中的数据准备好了，并且又再次收到了用户进程的system call，那么它马上就将数据拷贝到了用户内存，然后返回。

\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_nonblocking IO的特点是用户进程需要不断的主动询问kernel数据好了没有。

**I/O 多路复用（ IO multiplexing）**

IO multiplexing就是我们说的select，poll，epoll，有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select，poll，epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4][]

**`当用户进程调用了select，那么整个进程会被block`**，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。

这个图和blocking IO的图其实并没有太大的不同，事实上，还更差一些。因为这里需要使用两个system call (select 和 recvfrom)，而blocking IO只调用了一个system call (recvfrom)。但是，用select的优势在于它可以同时处理多个connection。

所以，如果处理的连接数不是很高的话，使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好，可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接。）

在IO multiplexing Model中，实际中，对于每一个socket，一般都设置成为non-blocking，但是，如上图所示，整个用户的process其实是一直被block的。只不过process是被select这个函数block，而不是被socket IO给block。

\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_\_I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符，而这些文件描述符（套接字描述符）其中的任意一个进入读就绪状态，select()函数就可以返回。

**异步 I/O（asynchronous IO）**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5][]

用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方面，从kernel的角度，当它受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了。

### 总结 ###

**blocking和non-blocking的区别**

调用blocking IO会一直block住对应的进程直到操作完成，而non-blocking IO在kernel还准备数据的情况下会立刻返回。

**synchronous IO和asynchronous IO的区别**

在说明synchronous IO和asynchronous IO的区别之前，需要先给出两者的定义。POSIX的定义是这样子的：  
\- A synchronous I/O operation causes the requesting process to be blocked until that I/O operation completes;  
\- An asynchronous I/O operation does not cause the requesting process to be blocked;

**两者的区别**就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义，之前所述的blocking IO，non-blocking IO，IO multiplexing都属于synchronous IO。

有人会说，non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方，定义中所指的”IO operation”是指真实的IO操作，就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候，如果kernel的数据没有准备好，这时候不会block进程。但是，当kernel中数据准备好的时候，recvfrom会将数据从kernel拷贝到用户内存中，这个时候进程是被block了，在这段时间内，进程是被block的。

而asynchronous IO则不一样，当进程发起IO 操作之后，就直接返回再也不理睬了，直到kernel发送一个信号，告诉进程说IO完成。在这整个过程中，进程完全没有被block。

各个IO Model的比较如图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 6][]

通过上面的图片，可以发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中，虽然进程大部分时间都不会被block，但是它仍然要求进程去主动的check，并且当数据准备完成以后，也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人（kernel）完成，然后他人做完后发信号通知。在此期间，用户进程不需要去检查IO操作的状态，也不需要主动的去拷贝数据。

### I/O 多路复用之select、poll、epoll详解 ###

select，poll，epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。

**select**

`select(rlist, wlist, xlist, timeout``=``None``)`

select 函数监视的文件描述符分3类，分别是writefds、readfds、和exceptfds。调用后select函数会阻塞，直到有描述副就绪（有数据 可读、可写、或者有except），或者超时（timeout指定等待时间，如果立即返回设为null即可），函数返回。当select函数返回后，可以 通过遍历fdset，来找到就绪的描述符。

select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点。select的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，可以通过修改宏定义甚至重新编译内核的方式提升这一限制，但 是这样也会造成效率的降低。  
  
**poll**

`int` `poll (struct pollfd ``*``fds, unsigned ``int` `nfds, ``int` `timeout);`

不同与select使用三个位图来表示三个fdset的方式，poll使用一个 pollfd的指针实现。

struct pollfd {
        int fd; /* file descriptor */
        short events; /* requested events to watch */
        short revents; /* returned events witnessed */
    };

pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式。同时，pollfd并没有最大数量限制（但是数量过大后性能也是会下降）。 和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符。

从上面看，select和poll都需要在返回后，`通过遍历文件描述符来获取已经就绪的socket`。事实上，同时连接的大量客户端在一时刻可能只有很少的处于就绪状态，因此随着监视的描述符数量的增长，其效率也会线性下降。

**epoll**

epoll是在2.6内核中提出的，是之前的select和poll的增强版本。相对于select和poll来说，epoll更加灵活，没有描述符限制。epoll使用一个文件描述符管理多个描述符，将用户关系的文件描述符的事件存放到内核的一个事件表中，这样在用户空间和内核空间的copy只需一次。

epoll操作过程

epoll操作过程需要三个接口，分别如下：

int epoll_create(int size)；//创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大
    int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；
    int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

1. int epoll\_create(int size);  
创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大，这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值，`参数size并不是限制了epoll所能监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议`。  
当创建好epoll句柄后，它就会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。

2. int epoll\_ctl(int epfd, int op, int fd, struct epoll\_event \*event)；  
函数是对指定描述符fd执行op操作。  
\- epfd：是epoll\_create()的返回值。  
\- op：表示op操作，用三个宏来表示：添加EPOLL\_CTL\_ADD，删除EPOLL\_CTL\_DEL，修改EPOLL\_CTL\_MOD。分别添加、删除和修改对fd的监听事件。  
\- fd：是需要监听的fd（文件描述符）  
\- epoll\_event：是告诉内核需要监听什么事

3. int epoll\_wait(int epfd, struct epoll\_event \* events, int maxevents, int timeout);  
等待epfd上的io事件，最多返回maxevents个事件。  
参数events用来从内核得到事件的集合，maxevents告之内核这个events有多大，这个maxevents的值不能大于创建epoll\_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回0表示已超时。

**Demo**

import socket, logging
    import select, errno
    
    logger = logging.getLogger("network-server")
    
    def InitLog():
        logger.setLevel(logging.DEBUG)
    
        fh = logging.FileHandler("network-server.log")
        fh.setLevel(logging.DEBUG)
        ch = logging.StreamHandler()
        ch.setLevel(logging.ERROR)
    
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        ch.setFormatter(formatter)
        fh.setFormatter(formatter)
    
        logger.addHandler(fh)
        logger.addHandler(ch)
    
    
    if __name__ == "__main__":
        InitLog()
    
        try:
            # 创建 TCP socket 作为监听 socket
            listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        except socket.error as  msg:
            logger.error("create socket failed")
    
        try:
            # 设置 SO_REUSEADDR 选项
            listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        except socket.error as  msg:
            logger.error("setsocketopt SO_REUSEADDR failed")
    
        try:
            # 进行 bind -- 此处未指定 ip 地址，即 bind 了全部网卡 ip 上
            listen_fd.bind(('', 2003))
        except socket.error as  msg:
            logger.error("bind failed")
    
        try:
            # 设置 listen 的 backlog 数
            listen_fd.listen(10)
        except socket.error as  msg:
            logger.error(msg)
    
        try:
            # 创建 epoll 句柄
            epoll_fd = select.epoll()
            # 向 epoll 句柄中注册 监听 socket 的 可读 事件
            epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
        except select.error as  msg:
            logger.error(msg)
    
        connections = {}
        addresses = {}
        datalist = {}
        while True:
            # epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
            epoll_list = epoll_fd.poll()
    
            for fd, events in epoll_list:
                # 若为监听 fd 被激活
                if fd == listen_fd.fileno():
                    # 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
                    conn, addr = listen_fd.accept()
                    logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
                    # 将连接 socket 设置为 非阻塞
                    conn.setblocking(0)
                    # 向 epoll 句柄中注册 连接 socket 的 可读 事件
                    epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
                    # 将 conn 和 addr 信息分别保存起来
                    connections[conn.fileno()] = conn
                    addresses[conn.fileno()] = addr
                elif select.EPOLLIN & events:
                    # 有 可读 事件激活
                    datas = ''
                    while True:
                        try:
                            # 从激活 fd 上 recv 10 字节数据
                            data = connections[fd].recv(10)
                            # 若当前没有接收到数据，并且之前的累计数据也没有
                            if not data and not datas:
                                # 从 epoll 句柄中移除该 连接 fd
                                epoll_fd.unregister(fd)
                                # server 侧主动关闭该 连接 fd
                                connections[fd].close()
                                logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
                                break
                            else:
                                # 将接收到的数据拼接保存在 datas 中
                                datas += data
                        except socket.error as  msg:
                            # 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
                            # 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
                            if msg.errno == errno.EAGAIN:
                                logger.debug("%s receive %s" % (fd, datas))
                                # 将已接收数据保存起来
                                datalist[fd] = datas
                                # 更新 epoll 句柄中连接d 注册事件为 可写
                                epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
                                break
                            else:
                                # 出错处理
                                epoll_fd.unregister(fd)
                                connections[fd].close()
                                logger.error(msg)
                                break
                elif select.EPOLLHUP & events:
                    # 有 HUP 事件激活
                    epoll_fd.unregister(fd)
                    connections[fd].close()
                    logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
                elif select.EPOLLOUT & events:
                    # 有 可写 事件激活
                    sendLen = 0
                    # 通过 while 循环确保将 buf 中的数据全部发送出去
                    while True:
                        # 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
                        sendLen += connections[fd].send(datalist[fd][sendLen:])
                        # 在全部发送完毕后退出 while 循环
                        if sendLen == len(datalist[fd]):
                            break
                    # 更新 epoll 句柄中连接 fd 注册事件为 可读
                    epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)
                else:
                    # 其他 epoll 事件不进行处理
                    continue

select 多并发socket 例子

Server

import select
    import socket
    import sys
    import queue
    
    
    server = socket.socket()
    server.setblocking(0)
    
    server_addr = ('localhost',10000)
    
    print('starting up on %s port %s' % server_addr)
    server.bind(server_addr)
    
    server.listen(5)
    
    
    inputs = [server, ] #自己也要监测呀,因为server本身也是个fd
    outputs = []
    
    message_queues = {}
    
    while True:
        print("waiting for next event...")
    
        readable, writeable, exeptional = select.select(inputs,outputs,inputs) #如果没有任何fd就绪,那程序就会一直阻塞在这里
    
        for s in readable: #每个s就是一个socket
    
            if s is server: #别忘记,上面我们server自己也当做一个fd放在了inputs列表里,传给了select,如果这个s是server,代表server这个fd就绪了,
                #就是有活动了, 什么情况下它才有活动? 当然 是有新连接进来的时候 呀
                #新连接进来了,接受这个连接
                conn, client_addr = s.accept()
                print("new connection from",client_addr)
                conn.setblocking(0)
                inputs.append(conn) #为了不阻塞整个程序,我们不会立刻在这里开始接收客户端发来的数据, 把它放到inputs里, 下一次loop时,这个新连接
                #就会被交给select去监听,如果这个连接的客户端发来了数据 ,那这个连接的fd在server端就会变成就续的,select就会把这个连接返回,返回到
                #readable 列表里,然后你就可以loop readable列表,取出这个连接,开始接收数据了, 下面就是这么干 的
    
                message_queues[conn] = queue.Queue() #接收到客户端的数据后,不立刻返回 ,暂存在队列里,以后发送
    
            else: #s不是server的话,那就只能是一个 与客户端建立的连接的fd了
                #客户端的数据过来了,在这接收
                data = s.recv(1024)
                if data:
                    print("收到来自[%s]的数据:" % s.getpeername()[0], data)
                    message_queues[s].put(data) #收到的数据先放到queue里,一会返回给客户端
                    if s not  in outputs:
                        outputs.append(s) #为了不影响处理与其它客户端的连接 , 这里不立刻返回数据给客户端
    
    
                else:#如果收不到data代表什么呢? 代表客户端断开了呀
                    print("客户端断开了",s)
    
                    if s in outputs:
                        outputs.remove(s) #清理已断开的连接
    
                    inputs.remove(s) #清理已断开的连接
    
                    del message_queues[s] ##清理已断开的连接
    
    
        for s in writeable:
            try :
                next_msg = message_queues[s].get_nowait()
    
            except queue.Empty:
                print("client [%s]" %s.getpeername()[0], "queue is empty..")
                outputs.remove(s)
    
            else:
                print("sending msg to [%s]"%s.getpeername()[0], next_msg)
                s.send(next_msg.upper())
    
    
        for s in exeptional:
            print("handling exception for ",s.getpeername())
            inputs.remove(s)
            if s in outputs:
                outputs.remove(s)
            s.close()
    
            del message_queues[s]

client

import socket
    import sys
    
    messages = [ b'This is the message. ',
                 b'It will be sent ',
                 b'in parts.',
                 ]
    server_address = ('localhost', 10000)
    
    # Create a TCP/IP socket
    socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              ]
    
    # Connect the socket to the port where the server is listening
    print('connecting to %s port %s' % server_address)
    for s in socks:
        s.connect(server_address)
    
    for message in messages:
    
        # Send messages on both sockets
        for s in socks:
            print('%s: sending "%s"' % (s.getsockname(), message) )
            s.send(message)
    
        # Read responses on both sockets
        for s in socks:
            data = s.recv(1024)
            print( '%s: received "%s"' % (s.getsockname(), data) )
            if not data:
                print(sys.stderr, 'closing socket', s.getsockname() )

**selectors模块**

This module allows high-level and efficient I/O multiplexing, built upon the [`select`][select] module primitives. Users are encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

import selectors
    import socket
     
    sel = selectors.DefaultSelector()
     
    def accept(sock, mask):
        conn, addr = sock.accept()  # Should be ready
        print('accepted', conn, 'from', addr)
        conn.setblocking(False)
        sel.register(conn, selectors.EVENT_READ, read)
     
    def read(conn, mask):
        data = conn.recv(1000)  # Should be ready
        if data:
            print('echoing', repr(data), 'to', conn)
            conn.send(data)  # Hope it won't block
        else:
            print('closing', conn)
            sel.unregister(conn)
            conn.close()
     
    sock = socket.socket()
    sock.bind(('localhost', 10000))
    sock.listen(100)
    sock.setblocking(False)
    sel.register(sock, selectors.EVENT_READ, accept)
     
    while True:
        events = sel.select()
        for key, mask in events:
            callback = key.data
            callback(key.fileobj, mask)

[Link 1]: http://baike.baidu.com/subview/121723/121723.htm
[Link 2]: http://baike.baidu.com/view/813.htm
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70]: /images/20220129/294cdbf7a1354be6a8cc119d2d0c19fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1]: /images/20220129/668b34c177994a50996fcbd15be0a21f.png
[Link 3]: http://guojing.me/linux-kernel-architecture/posts/process-switch/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2]: /images/20220129/75924873bfcd4723a2dcf09822103472.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3]: /images/20220129/f4d247d2b7304a3b9181ccd27d397ee1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4]: /images/20220129/b56d18dfa6c9474b904691f46d2a2ffe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5]: /images/20220129/fe881fed338e4ff38d34213dd2b8b683.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 6]: /images/20220129/96e94ba83aa4493da8b395852d0e893e.png
[select]: https://docs.python.org/3/library/select.html#module-select