标签：while import 地方 监视 类型 技术分享 blocking gen eve
select，poll，epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。
1、IO多路复用的三种方式,对比：
1、select--->效率最低，但有最大描述符限制，在linux为1024。
2、poll ---->和select一样，但没有最大描述符限制。
3、epoll --->效率最高，没有最大描述符限制，支持水平触发与边缘触发。
2、IO多路复用的优势：同时可以监听多个连接，用的是单线程，利用空闲时间实现并发。
3、三者之间的区别：
I/O多路复用：
1、select 的应用
select是通过系统调用来监视一组由多个文件描述符组成的数组，通过调用select()返回结果，数组中就绪的文件描述符会被内核标记出来，然后进程就可以获得这些文件描述符，然后进行相应的读写操作
select的实际执行过程如下：
所以对于我们监控的socket都要设置为非阻塞的，只有这样才能保证不会被阻塞。
优点:
缺点:
语法：r, w, e = select.select( rlist, wlist, errlist [,timeout] )
rlist,wlist和errlist均是waitable object； 都是文件描述符,就是一个整数，或者一个拥有返回文件描述符的函数fileno()的对象。解析如下：
在linux下这三个列表可以是空列表，但是在windows上不行
当超时时间没有设置时，如果监听的文件描述符没有任何变化，将会一直阻塞到发生变化为止。
当超时时间设置为1时，如果监听的描述符没有变化，则select会阻塞1秒，之后返回三个空列表。 如果由变化，则直接执行并返回。
举例如下：
服务端：（select 单线程支持多并发案例）
import select
import queue
import socket
server = socket.socket()
server.bind(("localhost",10002))
server.listen(5)
server.setblocking(False) #不阻塞
inputs = [server,]
outputs = []
msg_dic = {}
while True:
readable,writeable,exceptional = select.select(inputs,outputs,inputs)
#1、inputs让内核监测5个链接，如果有一个活动就返回，全部放到inputs。
#2、outputs是内核监测有异常的线程，
#3、inputs 处理异常的链接，看那些是断了，重新监听。
print(readable,writeable,exceptional)
for r in readable:
if r is server:#代表来了个新链接
conn,addr = server.accept()
print("来了新链接",addr)
inputs.append(conn) #是因为这个新建的链接还没发数据过来，现在接收程序就报错，所以想实现这个客户端发数据过来
#server端能知道，就需要让select再次监测
msg_dic[conn] = queue.Queue() #初始化一个队列，后面存要返回给这个客户端的数据。
else:
data = r.recv(1024)
print("收到数据",data)
msg_dic[r].put(data)
outputs.append(r) #放入返回的连接队列里
for w in writeable: #要返回给客户端的连接列表
data_to_client = msg_dic[w].get()
w.send(data_to_client)
outputs.remove(w) #确保下次循环的时候writeable，不返回已经处理完的连接
for e in exceptional: #删除数据，避免重复运行。
if e in outputs:
outputs.remove(e)
inputs.remove(e)
del msg_dic[e]
客户端：
import socket
#客户端
client = socket.socket() #声明socket 类型，同时生成socket链接对象
client.connect((‘127.0.0.1‘,10002))
while True:
msg = input(">>:").strip()
if len(msg) == 0:continue
client.send(msg.encode("utf-8"))
data = client.recv(10240)
print ("recv:",data.decode())
client.close()
2、poll的应用
poll本质上与select基本相同，只不过监控的最大连接数上相较于select没有了限制，因为poll使用的数据结构是链表，而select使用的是数组，数组是要初始化长度大小的，且不能改变。
poll原理：
poll状态：
优点：
缺点：
python使用poll方法：
举例：
服务端：
import select
import socket
import datetime
server_sock = socket.socket()
server_sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
# 需要设置socket选项时，需要先将socketlevel设置为SOL_SOCKET SOL=socket option level
# SO_REUSEADDR代表重用地址reuse addr
server_sock.bind(("localhost",10005))
server_sock.listen(5)
server_sock.setblocking(False) #设置为非阻塞
poll = select.poll()
poll.register(server_sock,select.POLLIN)
connections = {}
#遍历被监控的文件描述符
while True:
#print(datetime.datetime.now())
for fd,envent in poll.poll(100):
if envent == select.POLLIN:
if fd == server_sock.fileno():
#如果是当前sock，则接收请求
conn,addr = server_sock.accept()
poll.register(conn.fileno(),select.POLLIN)
connections[conn.fileno()] = conn
else:
conn = connections[fd]
data = conn.recv(1024)
if data:
print("%s accept %s" % (fd,data))
poll.modify(fd,select.POLLIN)
else:
conn = connections[fd]
try:
conn.send(b"Hello,%d" % fd)
print("conn>>:",conn)
finally:
poll.unregister(conn)
connections.pop(fd)
conn.close
客户端：
import socket
#客户端
client = socket.socket() #声明socket 类型，同时生成socket链接对象
client.connect((‘127.0.0.1‘,10005))
while True:
msg = input(">>:").strip()
if len(msg) == 0:continue
client.send(msg.encode("utf-8"))
data = client.recv(1024)
print ("recv:",data.decode())
client.close()
3、epoll的应用
epoll相当于是linux内核支持的方法，而epoll主要是解决select，poll的一些缺点
数组长度限制：
解决方案：fd上限是最大可以打开文件的数目，具体数目可以查看/proc/sys/fs/file-max。一般会和内存有关
需要每次轮询将数组全部拷贝到内核态：
解决方案：每次注册事件的时候，会把fd拷贝到内核态，而不是每次poll的时候拷贝，这样就保证每个fd只需要拷贝一次。
每次遍历都需要列表线性遍历：
解决方案：不再采用遍历的方案，给每个fd指定一个回调函数，fd就绪时，调用回调函数，这个回调函数会把fd加入到就绪的fd列表中，所以epoll只需要遍历就绪的list即可。
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);
A、int epoll_create(int size);
创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大，这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值，参数size并不是限制了epoll所能监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议。
当创建好epoll句柄后，它就会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。
B、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：是告诉内核需要监听什么事
C、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表示已超时。
举例：
服务端：
#_*_coding:utf-8_*_
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\poll\epoll 用得都比较少，后面会讲到一个模块：selectors 。selectors模块已经把 select、poll、epoll 进行封装。只有在做游戏开发的时候epoll用的才比较频繁。
IO多路复用中的两种触发方式：
水平触发:如果文件描述符已经就绪可以非阻塞的执行IO操作了,此时会触发通知.允许在任意时刻重复检测IO的状态, 没有必要每次描述符就绪后尽可能多的执行IO。select,poll就属于水平触发。
边缘触发:如果文件描述符自上次状态改变后有新的IO活动到来,此时会触发通知.在收到一个IO事件通知后要尽可能多的执行IO操作,因为如果在一次通知中没有执行完IO那么就需要等到下一次新的IO活动到来才能获取到就绪的描述符，信号驱动式IO就属于边缘触发。
epoll：即可以采用水平触发,也可以采用边缘触发。
1、水平触发：
只有高电频或低电频的时候才触发
举例：
服务端：
#水平触发
import socket
import select
sk=socket.socket()
sk.bind(("127.0.0.1",9904))
sk.listen(5)
while True:
r,w,e=select.select([sk,],[],[],5) #input输入列表，output输出列表，erron错误列表,5: 是监听5秒
for i in r: #[sk,]
print("hello")
print(‘>>>>>>‘)
客户端：
import socket
sk=socket.socket()
sk.connect(("127.0.0.1",9904))
while 1:
inp=input(">>").strip()
sk.send(inp.encode("utf8"))
data=sk.recv(1024)
print(data.decode("utf8"))
2、边缘触发
IO多路复用优势：同时可以监听多个连接
举例：select可以监控多个对象
服务端：
#优势
import socket
import select
sk=socket.socket()
sk.bind(("127.0.0.1",9904))
sk.listen(5)
inp=[sk,]
while True:
r,w,e=select.select(inp,[],[],5) #[sk,conn]，5是每隔几秒监听一次
for i in r: #[sk,]
conn,add=i.accept() #发送系统调用
print(conn)
print("hello")
inp.append(conn)
# conn.recv(1024)
print(‘>>>>>>‘)
客户端：
import socket
sk=socket.socket()
sk.connect(("127.0.0.1",9904))
while 1:
inp=input(">>").strip()
sk.send(inp.encode("utf8"))
data=sk.recv(1024)
print(data.decode("utf8"))
20、第七周-网络编程 - IO多路复用及select、poll、epoll模式详解
标签：while import 地方 监视 类型 技术分享 blocking gen eve
原文地址：https://www.cnblogs.com/chen170615/p/8932188.html
