day035协程、IO多路复用

本节内容：

1、协程（重点：gevent)
2、IO多路复用

一、协程

1、引子

本节的主题是基于单线程来实现并发，即只用一个主线程（很明显可利用的cpu只有一个）情况下实现并发，
为此我们需要先回顾下并发的本质：切换+保存状态

　　cpu正在运行一个任务，会在两种情况下切走去执行其他的任务（切换由操作系统强制控制），
一种情况是该任务发生了阻塞，另外一种情况是该任务计算的时间过长或有一个优先级更高的程序替代了它

　　协程本质上就是一个线程，以前线程任务的切换是由操作系统控制的，遇到I/O自动切换，
现在我们用协程的目的就是较少操作系统切换的开销（开关线程，创建寄存器、堆栈等，在他们之间进行切换等），
在我们自己的程序里面来控制任务的切换。

　　ps：在介绍进程理论时，提及进程的三种执行状态，而线程才是执行单位，所以也可以将上图理解为线程的三种状态

一：其中第二种情况并不能提升效率，只是为了让cpu能够雨露均沾，实现看起来所有任务都被“同时”执行的效果，

如果多个任务都是纯计算的，这种切换反而会降低效率。
为此我们可以基于yield来验证。
yield本身就是一种在单线程下可以保存任务运行状态的方法，我们来简单复习一下：

#1 yiled可以保存状态，yield的状态保存与操作系统的保存线程状态很像，但是yield是代码级别控制的，更轻量级
#2 send可以把一个函数的结果传给另外一个函数，以此实现单线程内程序之间的切换

通过yield实现任务切换+保存状态

import time

def func1():

    for i in range(11):
        #yield
        print('这是我第%s次打印啦' % i)
        time.sleep(1)

def func2():
    g = func1()
    #next(g)
    for k in range(10):

        print('哈哈，我第%s次打印了' % k)
        time.sleep(1)
        #next(g)

#不写yield，下面两个任务是执行完func1里面所有的程序才会执行func2里面的程序，有了yield，我们实现了两个任务的切换+保存状态
func1()
func2()

Python

Copy

单纯的切换反而会降低运行效率

#基于yield并发执行，多任务之间来回切换，这就是个简单的协程的体现，但是他能够节省I/O时间吗？不能
import time
def consumer():
    '''任务1:接收数据,处理数据'''
    while True:
        x=yield
        # time.sleep(1) #发现什么？只是进行了切换，但是并没有节省I/O时间
        print('处理了数据：',x)
def producer():
    '''任务2:生产数据'''
    g=consumer()
    next(g)  #找到了consumer函数的yield位置
    for i in range(3):
    # for i in range(10000000):
        g.send(i)  #给yield传值，然后再循环给下一个yield传值，并且多了切换的程序，比直接串行执行还多了一些步骤，导致执行效率反而更低了。
        print('发送了数据：',i)
start=time.time()
#基于yield保存状态,实现两个任务直接来回切换,即并发的效果
#PS:如果每个任务中都加上打印,那么明显地看到两个任务的打印是你一次我一次,即并发执行的.
producer() #我在当前线程中只执行了这个函数，但是通过这个函数里面的send切换了另外一个任务
stop=time.time()

# 串行执行的方式
# res=producer()
# consumer(res)
# stop=time.time()

print(stop-start)

Python

Copy

2、协程就是告诉Cpython解释器，你不是nb吗，不是搞了个GIL锁吗，

那好，我就自己搞成一个线程让你去执行，省去你切换线程的时间，我自己切换比你切换要快很多，避免了很多的开销，
对于单线程下，我们不可避免程序中出现io操作，但如果我们能在自己的程序中（即用户程序级别，而非操作系统级别）
控制单线程下的多个任务能在一个任务遇到io阻塞时就切换到另外一个任务去计算，
这样就保证了该线程能够最大限度地处于就绪态，即随时都可以被cpu执行的状态，
相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来，从而可以迷惑操作系统，
让其看到：该线程好像是一直在计算，io比较少，从而更多的将cpu的执行权限分配给我们的线程。

　　协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，
以此来提升效率。为了实现它，我们需要找寻一种可以同时满足以下条件的解决方案：

#1. 可以控制多个任务之间的切换，切换之前将任务的状态保存下来，以便重新运行时，可以基于暂停的位置继续执行。

#2. 作为1的补充：可以检测io操作，在遇到io操作的情况下才发生切换

3、协程介绍

协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。
一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。

对比操作系统控制线程的切换，用户在单线程内控制协程的切换
协程在操作系统上是没有这个概念的，是程序员们自己叫的

1.需要强调的是：

1. python的线程属于内核级别的，即由操作系统控制调度
（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他线程运行）

2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，
以此来提升效率（！！！非io操作的切换与效率无关）

2.优点如下：

1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu

3.缺点如下：

1. 协程的本质是单线程下，无法利用多核，
可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

4.总结协程特点：

1、必须在只有一个单线程里实现并发
2、修改共享数据不需加锁
3、用户程序里自己保存多个控制流的上下文栈
4、附加：一个协程遇到IO操作自动切换到其它协程
（如何实现检测IO，yield、greenlet都无法实现，就用到了gevent模块（select机制））

4、Greenlet

如果我们在单个线程内有20个任务，要想实现在多个任务之间切换，
使用yield生成器的方式过于麻烦（需要先得到初始化一次的生成器，然后再调用send。。。非常麻烦），
而使用greenlet模块可以非常简单地实现这20个任务直接的切换

#安装
pip3 install greenlet

1.生成器版的任务切换

import time

def func():
    for i in range(10):
        print("来这里")
        yield           # 通过yield记录上一次执行的位置
        time.sleep(1)
        print("%s号完成了" % i)

def func1():
    g = func()
    next(g)
    for i in range(10):
        time.sleep(1)
        print("%s号在忙" % i)
        next(g)               #

func1()

Python

Copy

2.效率对比

单纯的切换（在没有io的情况下或者没有重复开辟内存空间的操作），反而会降低程序的执行速度

效率对比

#顺序执行
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i

def f2():
    res=1
    for i in range(100000000):
        res*=i

start=time.time()
f1()
f2()
stop=time.time()
print('run time is %s' %(stop-start)) #10.985628366470337

#切换
from greenlet import greenlet
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i
        g2.switch()

def f2():
    res=1
    for i in range(100000000):
        res*=i
        g1.switch()

start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 52.763017892837524

Python

Copy

3.greenlet只是提供了一种便捷的切换方式，并没有提升效率

greenlet只是提供了一种比generator更加便捷的切换方式，
当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。

上面这个图，是协程真正的意义，虽然没有规避固有的I/O时间，但是我们使用这个时间来做别的事情了，
一般在工作中我们都是进程+线程+协程的方式来实现并发，以达到最好的并发效果，
如果是4核的cpu，一般起5个进程，每个进程中20个线程（5倍cpu数量），每个线程可以起500个协程，
大规模爬取页面的时候，等待网络延迟的时间的时候，我们就可以用协程去实现并发。
 并发数量 = 5 * 20 * 500 = 50000个并发，这是一般一个4cpu的机器最大的并发数。
 nginx在负载均衡的时候最大承载量就是5w个

单线程里的这20个任务的代码通常会既有计算操作又有阻塞操作，我们完全可以在执行任务1时遇到阻塞，
就利用阻塞的时间去执行任务2。。。。如此，才能提高效率，这就用到了Gevent模块。

Python

Copy

5、Gevent介绍

#安装
pip3 install gevent

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

1、用法

g1=gevent.spawn(func,1,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，
如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的，spawn是异步提交任务

g2=gevent.spawn(func2)

g1.join() #等待g1结束，上面只是创建协程对象，这个join才是去执行;如果不写会因为主代码太快，而没有被执行

g2.join() #等待g2结束  有人测试的时候会发现，不写第二个join也能执行g2，是的，协程帮你切换执行了，
但是你会发现，如果g2里面的任务执行的时间长，但是不写join的话，就不会执行完等到g2剩下的任务了


#或者上述两步合作一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

Python

Copy

2、gevent的简单使用

from gevent import monkey; monkey.patch_all()
import time
import gevent

def func1(n):
    print("xxxxx", n)
    gevent.sleep(2)
    # time.sleep(2) # 没有使用monkey,模块不能识别其他io切换，monkey模块的作用是标记出来
    print("eeeee", n)

def func2(m):
    print("11111", m)
    gevent.sleep(2)
    # time.sleep(2)
    print("222222", m)

start_time = time.time()
g1 = gevent.spawn(func1, "李白")
g2 = gevent.spawn(func2, "疏影")

# g1.join()
# g2.join()
gevent.joinall([g1,g2])  # 实现了单线程的并发（线程里面的协程），节省了时间

end_time = time.time()
print("运行时间", end_time - start_time)

print("主任务结束")
````
#### 上例gevent.sleep(2)模拟的是gevent可以识别的io阻塞,
    而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了

    from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面，如time，socket模块之前
    可以直接理解为：必须放在最文件开头，才能标记后面的io阻塞

    或者我们干脆记忆成：要用gevent，需要将from gevent import monkey;monkey.patch_all()放到文件的开头

```python
我们可以用threading.current_thread().getName()来查看每个g1和g2，
查看的结果为DummyThread-n，即假线程，虚拟线程，其实都在一个线程里面

进程线程的任务切换是由操作系统自行切换的，你自己不能控制

协程是通过自己的程序（代码）来进行切换的，自己能够控制，
只有遇到协程模块能够识别的IO操作的时候，程序才会进行任务切换，
实现并发效果，如果所有程序都没有IO操作，那么就基本属于串行执行了。

Python

Copy

3、 Gevent之同步与异步

协程：同步异步对比

from gevent import spawn,joinall,monkey;monkey.patch_all()

import time
def task(pid):
    """
    Some non-deterministic task
    """
    time.sleep(0.5)
    print('Task %s done' % pid)

def synchronous():
    for i in range(10):
        task(i)

def asynchronous():
    g_l=[spawn(task,i) for i in range(10)]
    joinall(g_l)

if __name__ == '__main__':
    print('Synchronous:')
    synchronous()

    print('Asynchronous:')
    asynchronous()
#上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。
初始化的greenlet列表存放在数组threads中，此数组被传给gevent.joinall 函数，
后者阻塞当前流程，并执行所有给定的greenlet。执行流程只会在 所有greenlet执行完后才会继续向下走。

Python

Copy

4、Gevent之应用举例一

协程应用：爬虫

from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time

def get_page(url):
    print('GET: %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        print('%d bytes received from %s' %(len(response.text),url))


start_time=time.time()
gevent.joinall([
    gevent.spawn(get_page,'https://www.python.org/'),
    gevent.spawn(get_page,'https://www.yahoo.com/'),
    gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))

Python

Copy

将上面的程序最后加上一段串行的代码看看效率：
如果你的程序不需要太高的效率，那就不用什么并发啊协程啊之类的东西。

print('--------------------------------')
s = time.time()
requests.get('https://www.python.org/')
requests.get('https://www.yahoo.com/')
requests.get('https://github.com/')
t = time.time()
print('串行时间>>',t-s)

Python

Copy

5、Gevent之应用举例二

通过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()
一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞）

一个网络请求里面经过多个时间延迟time

服务端

from gevent import monkey;monkey.patch_all()
from socket import *
import gevent

#如果不想用money.patch_all()打补丁,可以用gevent自带的socket
# from gevent import socket
# s=socket.socket()

def server(server_ip,port):
    s=socket(AF_INET,SOCK_STREAM)
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
    s.bind((server_ip,port))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    try:
        while True:
            res=conn.recv(1024)
            print('client %s:%s msg: %s' %(addr[0],addr[1],res))
            conn.send(res.upper())
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__ == '__main__':
    server('127.0.0.1',8080)

Python

Copy

客户端

from socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))


while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)

Python

Copy

多线程并发多个客户端，去请求上面的服务端是没问题的

from threading import Thread
from socket import *
import threading

def client(server_ip,port):
    c=socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，则大家公用一个套接字对象，那么客户端端口永远一样了
    c.connect((server_ip,port))

    count=0
    while True:
        c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
        msg=c.recv(1024)
        print(msg.decode('utf-8'))
        count+=1
if __name__ == '__main__':
    for i in range(500):
        t=Thread(target=client,args=('127.0.0.1',8080))
        t.start()

多线程并发多个客户端，去请求上面的服务端是没问题的

Python

Copy

二、IO多路复用

同步（synchronous） IO和异步（asynchronous） IO，阻塞（blocking） IO和非阻塞（non-blocking）IO分别是什么，
到底有什么区别？这个问题其实不同的人给出的答案都可能不同，
比如wiki，就认为asynchronous IO和non-blocking IO是一个东西。
这其实是因为不同的人的知识背景不同，并且在讨论这个问题的时候上下文(context)也不相同。
所以，为了更好的回答这个问题，我先限定一下本文的上下文。

　　本文讨论的背景是Linux环境下的network IO。
本文最重要的参考文献是Richard Stevens的“UNIX® Network Programming Volume 1, Third Edition: The Sockets Networking ”，
6.2节“I/O Models ”，Stevens在这节中详细说明了各种IO的特点和区别，
如果英文够好的话，推荐直接阅读。
Stevens的文风是有名的深入浅出，所以不用担心看不懂。本文中的流程图也是截取自参考文献。

1、Stevens在文章中一共比较了五种IO Model：

* blocking IO          阻塞IO

　　* nonblocking IO 非阻塞IO
　　* IO multiplexing IO多路复用
　　* signal driven IO 信号驱动IO（不常见，不讲）
　　* asynchronous IO 异步IO
　　由signal driven IO（信号驱动IO）在实际中并不常用，所以主要介绍其余四种IO Model。

　　再说一下IO发生时涉及的对象和步骤。对于一个network IO (这里我们以read、recv举例)，它会涉及到两个系统对象，一个是调用这个IO的process (or thread)，另一个就是系统内核(kernel)。当一个read/recv读数据的操作发生时，该操作会经历两个阶段：

1）等待数据准备 (Waiting for the data to be ready)
2）将数据从内核拷贝到进程中(Copying the data from the kernel to the process)

　　记住这两点很重要，因为这些IO模型的区别就是在两个阶段上各有不同的情况。

1.补充：

#1、输入操作：read、readv、recv、recvfrom、recvmsg共5个函数，
如果会阻塞状态，则会经理wait data和copy data两个阶段，如果设置为非阻塞则在wait 不到data时抛出异常

#2、输出操作：write、writev、send、sendto、sendmsg共5个函数，
在发送缓冲区满了会阻塞在原地，如果设置为非阻塞，则会抛出异常

#3、接收外来链接：accept，与输入操作类似

#4、发起外出链接：connect，与输出操作类似

Python

Copy

2、阻塞IO(blocking IO)

就是我们平常写的input,的代码，这样的阻塞
在linux中，默认情况下所有的socket都是blocking，
一个典型的读操作流程大概是这样：（recvfrom和tcp里面的recv在这些IO模型里面是一样的）

上面的图形分析：两个阶段的阻塞
所以，blocking IO的特点就是在IO执行的两个阶段（等待数据和拷贝数据两个阶段）都被block了。

　　这里我们回顾一下同步/异步/阻塞/非阻塞：

　　　　同步：提交一个任务之后要等待这个任务执行完毕

　　　　异步：只管提交任务，不等待这个任务执行完毕就可以去做其他的事情

　　　　阻塞：recv、recvfrom、accept，线程阶段 运行状态–>阻塞状态–>就绪

　　　　非阻塞：没有阻塞状态

在一个线程的IO模型中，我们recv的地方阻塞，我们就开启多线程，
但是不管你开启多少个线程，这个recv的时间是不是没有被规避掉，
不管是多线程还是多进程都没有规避掉这个IO时间。

3、非阻塞IO

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

在非阻塞式IO中，用户进程其实是需要不断的主动询问kernel数据准备好了没有。

虽然我们上面的代码通过设置非阻塞，规避了IO操作，但是非阻塞IO模型绝不被推荐。

非阻塞IO模型服务端

import time
import socket

server = socket.socket()

ip_port = ("127.0.0.1", 8001)
server.bind(ip_port)
server.listen()


server.setblocking(False)  # 设置不阻塞，把会阻塞的地方变得不阻塞，但是会报错，要用try捕获异常
conn_list = []  # 存放服务端对象，已经连接的管道对象

while 1:
    while 1:  # 不停的轮询内核，
        try:
            conn,addr = server.accept()  # 使用了setblocking后，这里不会阻塞，会有协程一直轮询内核，
            conn_list.append(conn)      # 将
            break
        except BlockingIOError:
            time.sleep(0.2)
            print("还没有接收到连接请求")

    print("你好")
    time.sleep(5)

    for sock in conn_list:
        print(sock)  # 管道对象
        while 1:
            try:
                from_cilent_msg = conn.recv(1024).decode("utf-8")
                print(from_cilent_msg)
                conn.send(b"hello")
                break
            except BlockingIOError:
                time.sleep(0.2)  # while循环很耗内存，用sleep缓冲一下
                print("没有接收到任何消息")

Python

Copy

非阻塞IO模型客户端

import socket

cilent = socket.socket()

ip_port = ("127.0.0.1", 8001)
cilent.connect(ip_port)

cilent_msg = input("请输入消息>>>>")
cilent.send(cilent_msg.encode("utf-8"))

from_server_msg = cilent.recv(1024).decode("utf-8")
print(from_server_msg)

Python

Copy

非阻塞IO示例详细版

# 服务端
import socket
import time

server=socket.socket()
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
server.bind(('127.0.0.1',8083))
server.listen(5)

server.setblocking(False) #设置不阻塞
r_list=[]  #用来存储所有来请求server端的conn连接
w_list={}  #用来存储所有已经有了请求数据的conn的请求数据

while 1:
    try:
        conn,addr=server.accept() #不阻塞，会报错
        r_list.append(conn)  #为了将连接保存起来，不然下次循环的时候，上一次的连接就没有了
    except BlockingIOError:
        # 强调强调强调：！！！非阻塞IO的精髓在于完全没有阻塞！！！
        # time.sleep(0.5) # 打开该行注释纯属为了方便查看效果
        print('在做其他的事情')
        print('rlist: ',len(r_list))
        print('wlist: ',len(w_list))


        # 遍历读列表，依次取出套接字读取内容
        del_rlist=[] #用来存储删除的conn连接
        for conn in r_list:
            try:
                data=conn.recv(1024) #不阻塞，会报错
                if not data: #当一个客户端暴力关闭的时候，会一直接收b''，别忘了判断一下数据
                    conn.close()
                    del_rlist.append(conn)
                    continue
                w_list[conn]=data.upper()
            except BlockingIOError: # 没有收成功，则继续检索下一个套接字的接收
                continue
            except ConnectionResetError: # 当前套接字出异常，则关闭，然后加入删除列表，等待被清除
                conn.close()
                del_rlist.append(conn)


        # 遍历写列表，依次取出套接字发送内容
        del_wlist=[]
        for conn,data in w_list.items():
            try:
                conn.send(data)
                del_wlist.append(conn)
            except BlockingIOError:
                continue


        # 清理无用的套接字,无需再监听它们的IO操作
        for conn in del_rlist:
            r_list.remove(conn)
        #del_rlist.clear() #清空列表中保存的已经删除的内容
        for conn in del_wlist:
            w_list.pop(conn)
        #del_wlist.clear()

#客户端
import socket
import os
import time
import threading
client=socket.socket()
client.connect(('127.0.0.1',8083))

while 1:
    res=('%s hello' %os.getpid()).encode('utf-8')
    client.send(res)
    data=client.recv(1024)

    print(data.decode('utf-8'))


##多线程的客户端请求版本
# def func():
#     sk = socket.socket()
#     sk.connect(('127.0.0.1',9000))
#     sk.send(b'hello')
#     time.sleep(1)
#     print(sk.recv(1024))
#     sk.close()
#
# for i in range(20):
#     threading.Thread(target=func).start()

Python

Copy

1.非阻塞IO模型绝不被推荐

虽然我们上面的代码通过设置非阻塞，规避了IO操作，但是非阻塞IO模型绝不被推荐。

　　我们不能否则其优点：能够在等待任务完成的时间里干其他活了
（包括提交其他任务，也就是 “后台” 可以有多个任务在“”同时“”执行）。

2.但是也难掩其缺点：

1. 循环调用recv()将大幅度推高CPU占用率；
这也是我们在代码中留一句time.sleep(2)的原因,否则在低配主机下极容易出现卡机情况

2. 任务完成的响应延迟增大了，因为每过一段时间才去轮询一次read操作，
而任务可能在两次轮询之间的任意时间完成。这会导致整体数据吞吐量的降低。

此外，在这个方案中recv()更多的是起到检测“操作是否完成”的作用，
实际操作系统提供了更为高效的检测“操作是否完成“作用的接口，
例如select()多路复用模式，可以一次检测多个连接是否活跃。

4、多路复用IO(IO multiplexing)（重点）

先看解释图，里面的select就像个代理。

IO multiplexing这个词可能有点陌生，但是如果我说select/epoll，大概就都能明白了。
有些地方也称这种IO方式为事件驱动IO(event driven IO)。
我们都知道，select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。
它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，
当某个socket有数据到达了，就通知用户进程。它的流程如图：

1.强调：

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

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

2.python中的select模块：

import select

fd_r_list, fd_w_list, fd_e_list = select.select(rlist, wlist, xlist, [timeout])

参数： 可接受四个参数（前三个必须）
    rlist: wait until ready for reading  #等待读的对象，你需要监听的需要获取数据的对象列表
    wlist: wait until ready for writing  #等待写的对象，你需要写一些内容的时候，input等等，
    也就是说我会循环他看看是否有需要发送的消息，如果有我取出这个对象的消息并发送出去，一般用不到，这里我们也给一个[]。

    xlist: wait for an “exceptional condition”  #等待异常的对象，一些额外的情况，一般用不到，但是必须传，那么我们就给他一个[]。
    timeout: 超时时间
    当超时时间 ＝ n(正整数)时，那么如果监听的句柄均无任何变化，则select会阻塞n秒，之后返回三个空列表，如果监听的句柄有变化，则直接执行。
返回值：三个列表与上面的三个参数列表是对应的
　　select方法用来监视文件描述符(当文件描述符条件不满足时，select会阻塞)，当某个文件描述符状态改变后，会返回三个列表
    1、当参数1 序列中的fd满足“可读”条件时，则获取发生变化的fd并添加到fd_r_list中
    2、当参数2 序列中含有fd时，则将该序列中所有的fd添加到 fd_w_list中
    3、当参数3 序列中的fd发生错误时，则将该发生错误的fd添加到 fd_e_list中
    4、当超时时间为空，则select会一直阻塞，直到监听的句柄发生变化

Python

Copy

3、结论:

select的优势在于可以处理多个连接，不适用于单个连接

fe1：io多路复用服务端、客户端示例

io多路复用服务端

import select
import socket

server = socket.socket()
server.bind(('127.0.0.1',8001))
rlist = [server,]  # 先将server对象放入列表，然后再将每个通道对象放入
server.listen()
while 1:
    print('11111')
    rl,wl,el = select.select(rlist,[],[]) # 每次会清空rl列表，重新监测
    print(222222)
    print('server对象>>>',server)
    print(rl)
    for sock in rl:  #  是server对象就添加通道对象，是通道，就行对应的收发消息
        if sock == server:
            conn,addr = sock.accept()
            rlist.append(conn)
        else:
            from_client_msg = sock.recv(1024)
            print(from_client_msg.decode('utf-8'))

# conn,addr = server.accept()
# from_client_msg = conn.recv(1024)
# print(from_client_msg.decode('utf-8'))

Python

Copy

io多路复用客户端

import socket

client = socket.socket()
client.connect(('127.0.0.1',8001))

to_server_msg = input('发给服务端的消息:')
client.send(to_server_msg.encode('utf-8'))
# from_server_msg = client.recv(1024)
# print(from_server_msg.decode('utf-8'))

Python

Copy

select网络IO模型的示例代码详细解释版

#服务端
from socket import *
import select
server = socket(AF_INET, SOCK_STREAM)
server.bind(('127.0.0.1',8093))
server.listen(5)
# 设置为非阻塞
server.setblocking(False)

# 初始化将服务端socket对象加入监听列表，后面还要动态添加一些conn连接对象，当accept的时候sk就有感应，当recv的时候conn就有动静
rlist=[server,]
rdata = {}  #存放客户端发送过来的消息

wlist=[]  #等待写对象
wdata={}  #存放要返回给客户端的消息

print('预备！监听！！！')
count = 0 #写着计数用的，为了看实验效果用的，没用
while True:
    # 开始 select 监听,对rlist中的服务端server进行监听，select函数阻塞进程，直到rlist中的套接字被触发（在此例中，套接字接收到客户端发来的握手信号，从而变得可读，满足select函数的“可读”条件），被触发的（有动静的）套接字（服务器套接字）返回给了rl这个返回值里面；
    rl,wl,xl=select.select(rlist,wlist,[],0.5)
    print('%s 次数>>'%(count),wl)
    count = count + 1
    # 对rl进行循环判断是否有客户端连接进来,当有客户端连接进来时select将触发
    for sock in rl:
        # 判断当前触发的是不是socket对象, 当触发的对象是socket对象时,说明有新客户端accept连接进来了
        if sock == server:
            # 接收客户端的连接, 获取客户端对象和客户端地址信息
            conn,addr=sock.accept()
            #把新的客户端连接加入到监听列表中，当客户端的连接有接收消息的时候，select将被触发，会知道这个连接有动静，有消息，那么返回给rl这个返回值列表里面。
            rlist.append(conn)
        else:
            # 由于客户端连接进来时socket接收客户端连接请求，将客户端连接加入到了监听列表中(rlist)，客户端发送消息的时候这个连接将触发
            # 所以判断是否是客户端连接对象触发
            try:
                data=sock.recv(1024)
                #没有数据的时候，我们将这个连接关闭掉，并从监听列表中移除
                if not data:
                    sock.close()
                    rlist.remove(sock)
                    continue
                print("received {0} from client {1}".format(data.decode(), sock))
                #将接受到的客户端的消息保存下来
                rdata[sock] = data.decode()

                #将客户端连接对象和这个对象接收到的消息加工成返回消息，并添加到wdata这个字典里面
                wdata[sock]=data.upper()
                #需要给这个客户端回复消息的时候，我们将这个连接添加到wlist写监听列表中
                wlist.append(sock)
            #如果这个连接出错了，客户端暴力断开了（注意，我还没有接收他的消息，或者接收他的消息的过程中出错了）
            except Exception:
                #关闭这个连接
                sock.close()
                #在监听列表中将他移除，因为不管什么原因，它毕竟是断开了，没必要再监听它了
                rlist.remove(sock)
    # 如果现在没有客户端请求连接,也没有客户端发送消息时，开始对发送消息列表进行处理，是否需要发送消息
    for sock in wl:
        sock.send(wdata[sock])
        wlist.remove(sock)
        wdata.pop(sock)

    # #将一次select监听列表中有接收数据的conn对象所接收到的消息打印一下
    # for k,v in rdata.items():
    #     print(k,'发来的消息是：',v)
    # #清空接收到的消息
    # rdata.clear()

---------------------------------------
#客户端
from socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8093))


while True:
    msg=input('>>: ').strip()
    if not msg:continue
    client.send(msg.encode('utf-8'))
    data=client.recv(1024)
    print(data.decode('utf-8'))

client.close()

Python

Copy

4、select监听fd变化的过程分析：

用户进程创建socket对象，拷贝监听的fd到内核空间，每一个fd会对应一张系统文件表，
内核空间的fd响应到数据后，就会发送信号给用户进程数据已到；

用户进程再发送系统调用，比如（accept）将内核空间的数据copy到用户空间，同时作为接受数据端内核空间的数据清除，
这样重新监听时fd再有新的数据又可以响应到了（发送端因为基于TCP协议所以需要收到应答后才会清除）。

5、该模型的优点：

相比其他模型，使用select() 的事件驱动模型只用单线程（进程）执行，占用资源少，不消耗太多 CPU，
同时能够为多客户端提供服务。如果试图建立一个简单的事件驱动的服务器程序，这个模型有一定的参考价值。

6、该模型的缺点：

首先select()接口并不是实现“事件驱动”的最好选择。因为当需要探测的句柄值较大时，
select()接口本身需要消耗大量时间去轮询各个句柄。很多操作系统提供了更为高效的接口，
如linux提供了epoll，BSD提供了kqueue，Solaris提供了/dev/poll，…。如果需要实现更高效的服务器程序，
类似epoll这样的接口更被推荐。遗憾的是不同的操作系统特供的epoll接口有很大差异，
所以使用类似于epoll的接口实现具有较好跨平台能力的服务器会比较困难。

其次，该模型将事件探测和事件响应夹杂在一起，一旦事件响应的执行体庞大，则对整个模型是灾难性的。

7、IO多路复用的机制：

Select  将所有需要监听的对象,放到一个列表里面,将这个列表交给select来监听,
凡是有动静的对象,直接给你返回到一个列表中,然后我们循环这个列表,根据列表里面的有动静的对象,来进行对应的操作.  window 和 linux

Poll  : 监听的对象数量没有上限,而select默认是1024个,Linux
    Select和poll内部在监听的所有对象的时候,是循环遍历的操作,挨个问一遍有没有数据

Epoll : linux,每个被监听的对象,都是通过回调机制来搞的,
也就是,监听的列表对象里面的这个对象,凡是有动静,自行通知epoll,然后epoll把这些有动静的对象返回.

5、异步IO(Asynchronous I/O)

最NB的,效率最高,因为将网络请求里面的两个io时间都省了