python并发编程-IO多路复用select/poll/epoll实现多客户端通信

左手的ㄟ右手 2021-11-02 08:14 408阅读 0赞

# python并发编程-IO多路复用select/poll/epoll实现多客户端通信 #

### 定义 ###

> 同时监控多个IO事件，当哪个IO事件准备就绪就（已经到了必然要处理的步骤）执行哪个IO事件。以此形成可以同时处理多个IO的行为，避免一个IO阻塞造成其他IO均无法执行，提高了IO执行效率。

### 具体方案 ###

> select方法: Windows/Linux/Unix  
> poll方法: Linux/Unix  
> epoll方法: Linux

### select方法 ###

> rs, ws, xs=select(rlist, wlist, xlist\[, timeout\])  
> **功能:**  
> 监控IO事件，阻塞等待IO发生  
> **参数：**  
> rlist 列表 存放关注的等待发生的IO事件　比如：accept()等待客户端连接  
> wlist 列表 存放关注的要主动处理的IO事件　比如：sendto()发消息  
> xlist 列表 存放关注的出现异常要处理的IO　比如：IO发生异常  
> timeout 超时时间  
> **返回值：**  
> rs 列表 rlist中准备就绪的IO  
> ws 列表 wlist中准备就绪的IO  
> xs 列表 xlist中准备就绪的IO  
> 这三个返回列表，至少有一个不为空

### 注意 ###

> wlist中如果存在IO事件，则select立即返回给ws处理IO过程中不要出现死循环占有服务端的情况IO多路复用消耗资源较少，效率较高同时监听1024个IO事件

### select 实现tcp服务 ###

> 【1】 将关注的IO放入对应的监控类别列表  
> 【2】通过select函数进行监控  
> 【3】遍历select返回值列表，确定就绪IO事件  
> 【4】处理发生的IO事件

### select实现多客户端通信 ###

服务端代码

from socket import *
    from select import select
    
    # 　设置套接字为关注IO
    s = socket()
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.bind(("0.0.0.0", 9548))
    s.listen(5)
    
    #  设置关注的IO
    rlist = [s]
    wlist = []
    xlist = []
    
    # 循环监控IO的发生
    while True:
        print("开始监控IO行为")
        rs, ws, xs = select(rlist, wlist, xlist)
        #  遍历三个返回值列表,进一步判断哪个IO发生
        for r in rs:
            #  如果是套接字就绪,就处理连接
            if r is s:
                c, addr = r.accept()
                print("Connect from:", addr)
                rlist.append(c)  # 加入新的关注IO
            else:
                data = r.recv(1024)
                if not data:
                    rlist.remove(r)
                    r.close()
                    continue
                print(data.decode())
                # r.send(b"OK")
                # 希望我们主动处理这个IO
                wlist.append(r)
    
        for w in ws:
            w.send(b"OK Thanks")
            wlist.remove(w)
        for x in xs:
            pass

客户端代码

from socket import *
    
    sockfd = socket(AF_INET, SOL_SOCKET)
    
    sockfd.connect(("172.40.71.158", 9548))
    while True:
        msg = input("输入消息：")
        sockfd.send(msg.encode())
        data = sockfd.recv(1024)
        if not data:
            print("服务器已退出")
            break
        print("接收到的消息为:", data.decode())
    
    sockfd.close()

### poll方法 ###

> p = select.poll()　＃这个poll是创建对象  
> **功能 ：**  
> 创建poll对象  
> **返回值：**  
> poll对象  
> p.register(fd,event)  
> **功能:**  
> 注册关注的IO事件  
> **参数：**  
> fd 要关注的IO  
> event 要关注的IO事件类型  
> 监测的事件类型按照按位或进行操作,第一个参数为监测事件IO,第二个参数为需要监测事件类型  
> **常用类型：**  
> POLLIN 读IO事件（rlist）  
> POLLOUT 写IO事件 (wlist)  
> POLLERR 异常IO （xlist）  
> POLLHUP 断开连接  
> e.g. p.register(sockfd,POLLIN|POLLERR)  
> p.unregister(fd)  
> **功能：**  
> 取消对IO的关注  
> **参数：**  
> IO对象或者IO对象的fileno  
> events = p.poll()　＃这个poll()是阻塞监控  
> **功能：**  
> 阻塞等待监控的IO事件发生  
> **返回值：**  
> 返回发生的IO  
> events格式 \[(fileno,event),()…\]  
> 每个元组为一个就绪IO，元组第一项是该IO的fileno，第二项为该IO就绪的事件类型。只能根据文件描述符回推那个阻塞对象。建立查找地图

### poll实现tcp服务 ###

> 【1】 创建套接字  
> 【2】 将套接字register  
> 【3】 建立查找地图，创建查找字典，并维护  
> 【4】 循环监控IO发生  
> 【5】 处理发生的IO

### poll实现多客户端通信 ###

服务端代码

from select import *
    from socket import *
    
    sockfd = socket()
    sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sockfd.bind(("0.0.0.0", 9548))
    sockfd.listen(5)
    
    #  创建poll对象
    p = poll()
    
    #  建立一个查找字典(地图){fileno:io_obj}
    fdmap = {sockfd.fileno(): sockfd}
    
    #  设置关注IO
    p.register(sockfd, POLLIN | POLLERR)
    
    #  循环监控IO时间发生
    while True:
        events = p.poll()  # 阻塞等待IO发生
        #  遍历列表处理IO
        for fd, event in events:  # fd为文件描述符,event为IO就绪事件类型
            if fd == sockfd.fileno():
                c, addr = fdmap[fd].accept()
                print("Connect from:", addr)
                #  添加新的关注事件
                p.register(c, POLLIN | POLLHUP)
                fdmap[c.fileno()] = c
            elif event & POLLIN:  # 客户端发消息
                data = fdmap[fd].recv(1024)
                if not data:
                    print("客户端退出")
                    p.unregister(fd)  # 取消关注事件
                    fdmap[fd].close()
                    del fdmap[fd]  # 从字典删除
                    continue
                print(data.decode())
                fdmap[fd].send(b"OK")

客户端代码同select客户端代码

### epoll方法 ###

**使用方法**

> 基本与poll相同  
> 生成对象改为 epoll()  
> 将所有事件类型改为EPOLL类型

**epoll特点（效率是最高的）**

> **epoll 效率比select poll要高**  
> select poll将需要监听的IO事件交给内核,当有可以执行的IO事件发生时,内核会将所有的IO事件交给应用层,包括未准备就绪的IO事件,应用层需要遍历拿到准备就绪的IO事件,影响效率.而epoll是在内核开辟一块空间,哪个IO事件准备就绪,就将哪个IO事件交给应用层,效率得以提升.

> **epoll 监控IO数量比select poll要多**  
> select poll同时监听1024个IO事件,而epoll在1GB内存的机器上大约是10万左右.

> **epoll 的触发方式比poll要多 （EPOLLET边缘触发）epoll默认是水平触发**  
> 水平触发,当一个IO事件准备就绪,应用层未去做处理,它会一直通知你,而边缘触发不会,第一次未处理它不会再去通知,当再次接收到消息时,第二次会将第一次通知的消息一起带过去.

### epoll实现多客户端通信 ###

服务端代码

from select import *
    from socket import *
    
    sockfd = socket()
    sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sockfd.bind(("0.0.0.0", 9549))
    sockfd.listen(5)
    
    #  创建epoll对象
    ep = epoll()
    #  建立一个查找字典(地图){fileno:io_obj}
    fdmap = {sockfd.fileno(): sockfd}
    
    #  设置关注IO
    ep.register(sockfd, EPOLLIN | EPOLLERR)
    
    #  循环监控IO时间发生
    while True:
        events = ep.poll()  # 阻塞等待IO发生
        #  遍历列表处理IO
        for fd, event in events:  # fd为文件描述符,event为IO就绪事件类型
            if fd == sockfd.fileno():
                c, addr = fdmap[fd].accept()
                print("Connect from:", addr)
                #  添加新的关注事件
                ep.register(c, EPOLLIN | EPOLLHUP)
                fdmap[c.fileno()] = c
            elif event & EPOLLIN:  # 客户端发消息
                data = fdmap[fd].recv(1024)
                if not data:
                    print("客户端退出")
                    ep.unregister(fd)  # 取消关注事件
                    fdmap[fd].close()
                    del fdmap[fd]  # 从字典删除
                    continue
                print(data.decode())
                fdmap[fd].send(b"OK")

客户端代码同select客户端代码