论事件驱动与异步IO
事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是（单线程）同步以及多线程编程。
让我们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展示了随着时间的推移，这三种模式下程序所做的工作。这个程序有3个任务需要完成，每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。
在单线程同步模型中，任务按照顺序执行。如果某个任务因为I/O而阻塞，其他所有的任务都必须等待，直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系，但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。
在多线程版本中，这3个任务分别在独立的线程中执行。这些线程由操作系统来管理，在多处理器系统上可以并行处理，或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。与完成类似功能的同步程序相比，这种方式更有效率，但程序员必须写代码来保护共享资源，防止其被多个线程同时访问。多线程程序更加难以推断，因为这类程序不得不通过线程同步机制如锁、可重入函数、线程局部存储或者其他机制来处理线程安全问题，如果实现不当就会导致出现微妙且令人痛不欲生的bug。
在事件驱动版本的程序中，3个任务交错执行，但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时，注册一个回调到事件循环中，然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件，当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽可能的得以执行而不需要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为，因为程序员不需要关心线程安全问题。
当我们面对如下的环境时，事件驱动模型通常是一个好的选择：
- 程序中有许多任务，而且…
- 任务之间高度独立（因此它们不需要互相通信，或者等待彼此）而且…
- 在等待事件到来时，某些任务会阻塞。
当应用程序需要在任务间共享可变的数据时，这也是一个不错的选择，因为这里不需要采用同步处理。
网络应用程序通常都有上述这些特点，这使得它们能够很好的契合事件驱动编程模型。
Select\Poll\Epoll异步IO
Python Select 解析
首先列一下，sellect、poll、epoll三者的区别
select
select最早于1983年出现在4.2BSD中，它通过一个“select()系统”调用 来监控多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。
（也可以这么解释：这个模型中一些模块配置的是非阻塞I/O,然后使用阻塞select系统调用来确定一个I/O描述符何时有操作。使用select调用可以为多个描述符提供通知，对于每个提示符，我们可以请求描述符的可写，可读以及是否发生错误。）
select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点，事实上从现在看来，这也是它所剩不多的优点之一。
select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。
另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。
复制开销线性增长是怎么回事?
操作系统维护"高速的事件循环的队列",程序在系统中注册一个事件后,程序需要隔段时间自己去取结果.这时候操作系统会把所有注册的任务列表给到程序.返回之后,1000个任务,程序自己循环,找到自己的任务.那么为什么说随着文件描述符的增大,复制的开销也线程增大?
首先我们知道在Linux系统中每一个用户\每一个进程能够打开文件的数量默认是1024,也就是前面说的Linux中单个进程能够监视的文件喵舒服的数量默认限制在1024内.当然这个可以通过更改/etc/security/limits.conf配置文件来调整.
当一个进程中对两个文件进行读写,每一个读写的需求就会在操作系统的高速循环事件的数组中注册一个任务,那么这个任务提交后操作系统进行io操作,接下来进程会拿着之前的注册信息隔段时间去问下操作系统,我的任务执行完了没?操作系统作为一个公务员,才不会浪费宝贵的时间,而是高速他所有的信息都是自助查询的.由于政府资源有限 ,你从这边只能拷贝这些信息,然后在你自己的电脑上查吧.
所以进程就得把操作系统的 "高速循环事件任务数组"拷贝过来,在根据自己的注册号,对这1000个任务或者更多去for循环对比.直到找到自己的号,看看状态是不是已经就绪了.如果就绪了就把数据取下来,如果没就绪,只能隔段时间再去问拷贝.当有多个IO操作时,可想而之,这个拷贝的量级当然是线性增长.同时内核的"高速事件循环数组"的大小可能是百万级的.
所以说select的效率是不高的. poll
poll在1986年诞生于System V Release 3，它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。 poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。 另外，select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll()的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。 epoll
直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。 epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。 epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。 另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。
epoll和程序怎样交互的?
当程序有IO操作时,向操作系统内核注册一个IO任务,并添加到"高速事件循环任务数组".epoll在接收进程的注册一个文件描述符请求后,首先通过epoll_ctl()来注册一个文件描述符,和select()不同的是,他不需要进程调用select()里相应的方法后才对所有监控的文件描述符进行扫描.而是一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符.
而当有进程调用epoll_wait()方法后,epoll会将已经就绪的文件描述符的映射(注意是映射,而不是所有的数组).这个映射相当于文件句柄,进程可以直接对这个映射循环,当循环到自己注册的文件描述符后,通过这个映射就可以进行读写IO操作了.
这里面还有一个问题?epoll回调机制,激活文件描述符后,只有当相关进程调用epoll_wait()方法才会得到通知,而当进程在执行其他代码,没有立刻执行epoll_wait()方法时,操作系统又不能停掉进程现在的动作来执行epoll_wait(),这时操作系统会怎么处理这个文件描述符呢?操作系统会将这个文件描述符,放到一个专门存放就绪文件描述符的列表里.等着进程下一次epoll_wait()的时候,通过映射返回给进程.
这个时候,操作系统已经把这个数据缓存在内存里,但这个缓存不会一直缓存,所以有一个缓冲区,可能几十K,如果这个缓冲区满了,它就不接IO操作的文件描述符注册请求了,内核会告诉进程你等会在发申请,我这边的缓冲区满了.那么会产生什么效果呢?进程就会阻塞住了.等到进程调用epoll_wait()时,缓存空间减少,这时候其他进程就可以正常使用了.
总结: epoll 相对与 select优化了3大点:1.返回给进程的是已就绪的文件描述符的映射,优化了复制开销. 2.通知的方式由进程触发系统扫描,epoll优化成文件描述符一旦就绪,内核采用回调机制,激活这个文件描述符,这样当相关进程在调用epoll_wait()时,便得到通知.(我觉得如果真是这样,进程只会在得到通知后才会for循环已就绪的文件描述符的映射,相当于节省了很多没必要的循环.)
epoll是最流行的一种多路IO异步复用的模型.
上面的知识理解即可,因为没有人面试的时候让你写epoll代码,但是原理要知道,才能做一个牛逼的开发. Python select Python的select()方法直接调用操作系统的IO接口，它监控sockets,open files, and pipes(所有带fileno()方法的文件句柄)何时变成readable 和writeable, 或者通信错误，select()使得同时监控多个连接变的简单，并且这比写一个长循环来等待和监控多客户端连接要高效，因为select直接通过操作系统提供的C的网络接口进行操作，而不是通过Python的解释器。 接下来通过echo server例子(就是接收socket客户端发过来的数据并发送给客户端的例子)要以了解select 是如何通过单进程实现同时处理多个非阻塞的socket连接的
#!/usr/bin/env python3.
#__author__:'ted.zhou'
'''
使用select实现异步IO
''' import select
import socket
import sys
import queue # Create a TCP/IP socket
server = socket.socket(socket.AF_INET,socket.SOCK_STREAM) # 创建一个socket实例,定义地址簇为IPV4,socket类型是TCP
server.setblocking() # 设置server这个socket实例为非阻塞
# 正是因为我们要掩饰select实现异步IO模型,所以才将所有的阻塞操作交由select来控制 # bind the socket to the port server_address = ('localhost',)
print >> sys.stderr, 'starting up on %s port %s' % server_address
server.bind(server_address) # Listen for incoming connections
server.listen()
select()方法接收并监控3个通信列表， 第一个是所有的输入的data,就是指外部发过来的数据，第2个是监控和接收所有要发出去的data(outgoing data),第3个监控错误信息，接下来我们需要创建2个列表来包含输入和输出信息来传给select().
# Sockets from which we expect to read
inputs = [ server ] # Sockets to which we expect to write
outputs = [ ]
所有客户端的进来的连接和数据将会被server的主循环程序放在上面的list中处理，我们现在的server端需要等待连接可写（writable)之后才能过来，然后接收数据并返回(因此不是在接收到数据之后就立刻返回)，因为每个连接要把输入或输出的数据先缓存到queue里，然后再由select取出来再发出去。
# Outgoing message queues (socket:Queue)
message_queues = {}
下面是此程序的主循环，调用select()时会阻塞和等待直到新的连接和数据进来
while inputs: # Wait for at least one of the sockets to be ready for processing
print >>sys.stderr, '\nwaiting for the next event'
readable, writable, exceptional = select.select(inputs, outputs, inputs)
当你把inputs,outputs,exceptional(这里跟inputs共用)传给select()后，它返回3个新的list，我们上面将他们分别赋值为readable,writable,exceptional, 所有在readable list中的socket连接代表有数据可接收(recv),所有在writable list中的存放着你可以对其进行发送(send)操作的socket连接，当连接通信出现error时会把error写到exceptional列表中。
Readable list 中的socket 可以有3种可能状态，第一种是如果这个socket是main "server" socket,它负责监听客户端的连接，如果这个main server socket出现在readable里，那代表这是server端已经ready来接收一个新的连接进来了，为了让这个main server能同时处理多个连接，在下面的代码里，我们把这个main server的socket设置为非阻塞模式。
# Handle inputs
for s in readable: if s is server:
# A "readable" server socket is ready to accept a connection
connection, client_address = s.accept()
print >>sys.stderr, 'new connection from', client_address
connection.setblocking()
inputs.append(connection) # Give the connection a queue for data we want to send
message_queues[connection] = Queue.Queue()
第二种情况是这个socket是已经建立了的连接，它把数据发了过来，这个时候你就可以通过recv()来接收它发过来的数据，然后把接收到的数据放到queue里，这样你就可以把接收到的数据再传回给客户端了。
else:
data = s.recv()
if data:
# A readable client socket has data
print >>sys.stderr, 'received "%s" from %s' % (data, s.getpeername())
message_queues[s].put(data)
# Add output channel for response
if s not in outputs:
outputs.append(s)
第三种情况就是这个客户端已经断开了，所以你再通过recv()接收到的数据就为空了，所以这个时候你就可以把这个跟客户端的连接关闭了。
else:
# Interpret empty result as closed connection
print >>sys.stderr, 'closing', client_address, 'after reading no data'
# Stop listening for input on the connection
if s in outputs:
outputs.remove(s) #既然客户端都断开了，我就不用再给它返回数据了，所以这时候如果这个客户端的连接对象还在outputs列表中，就把它删掉
inputs.remove(s) #inputs中也删除掉
s.close() #把这个连接关闭掉 # Remove message queue
del message_queues[s]
对于writable list中的socket，也有几种状态，如果这个客户端连接在跟它对应的queue里有数据，就把这个数据取出来再发回给这个客户端，否则就把这个连接从output list中移除，这样下一次循环select()调用时检测到outputs list中没有这个连接，那就会认为这个连接还处于非活动状态
# Handle outputs
for s in writable:
try:
next_msg = message_queues[s].get_nowait()
except Queue.Empty:
# No messages waiting so stop checking for writability.
print >>sys.stderr, 'output queue for', s.getpeername(), 'is empty'
outputs.remove(s)
else:
print >>sys.stderr, 'sending "%s" to %s' % (next_msg, s.getpeername())
s.send(next_msg)
最后，如果在跟某个socket连接通信过程中出了错误，就把这个连接对象在inputs\outputs\message_queue中都删除，再把连接关闭掉
# Handle "exceptional conditions"
for s in exceptional:
print >>sys.stderr, 'handling exceptional condition for', s.getpeername()
# Stop listening for input on the connection
inputs.remove(s)
if s in outputs:
outputs.remove(s)
s.close() # Remove message queue
del message_queues[s]
客户端
下面的这个是客户端程序展示了如何通过select()对socket进行管理并与多个连接同时进行交互，
import socket
import sys messages = [ 'This is the message. ',
'It will be sent ',
'in parts.',
]
server_address = ('localhost', ) # 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 >>sys.stderr, 'connecting to %s port %s' % server_address
for s in socks:
s.connect(server_address)
接下来通过循环通过每个socket连接给server发送和接收数据。
for message in messages: # Send messages on both sockets
for s in socks:
print >>sys.stderr, '%s: sending "%s"' % (s.getsockname(), message)
s.send(message) # Read responses on both sockets
for s in socks:
data = s.recv()
print >>sys.stderr, '%s: received "%s"' % (s.getsockname(), data)
if not data:
print >>sys.stderr, 'closing socket', s.getsockname()
实际使用代码
server 端：
#!/usr/bin/env python3.
#__author__:'ted.zhou'
'''
使用select实现异步IO
''' import select
import socket
import sys
import queue # Create a TCP/IP socket
server = socket.socket(socket.AF_INET,socket.SOCK_STREAM) # 创建一个socket实例,定义地址簇为IPV4,socket类型是TCP
server.setblocking() # 设置server这个socket实例为非阻塞
# 正是因为我们要掩饰select实现异步IO模型,所以才将所有的阻塞操作交由select来控制 # bind the socket to the port server_address = ('localhost',) # 指定socket启动绑定的IP和port
print(sys.stderr, 'starting up on %s port %s' % server_address) # 打印开始启动socket 的IP和port
server.bind(server_address) # 把IP_Port 绑定到server实例 # Listen for incoming connections
server.listen() # 启动server实例,并设置最大等待连接数为5个 # Sockets from which we expect to read
inputs = [ server ] # 创建一个inputs列表,用来存储server实例 和 进入的新连接实例,主要作用是分三种:.server接收新链接. .已创建的连接接收数据 .已存在的连接接收到客户端关闭的信息,以此删除这个旧连接
# 这里要注意下一个概念,socket创建出来的实例,每一个实例的方法对于select()方法来说,实例只要调用了任何方法都是有所改变的.
# 老师说这里为什么要把server加入到inputs列表中,除了要使下面的代码循环下去,还有其他作用.但是我觉得这里其实就是为了让select监控server这个实例,就是这么简单,监控server实例的变化 # Sockets to which we expect to write
outputs = [ ] # 创建select监控的列表2,用于发送数据
message_queues = {} # 创建一个字典,字典用来保存每一个连接的队列. 键值对为 连接实例:队列
account = # 循环计数,从计数中我们可以看出select()方法,在什么时候触发.经测试为发现select对不同列表的监控规则是不一样的. while inputs:
print(account)
# Wait for at least one of the sockets to be ready for processing
print( '\nwaiting for the next event')
readable, writable, exceptional = select.select(inputs, outputs, inputs)
'''
经测试为发现select对不同列表的监控规则是不一样的.
为什么说select()对它所监控的三个列表的监控规则不一样的呢?我先把测试出来的结果列出来:
.对inputs列表只监控列表中的元素的修改状态(也就是监控元素调用的任何方法都算状态改变),不监控列表中新加的元素和删除的元素,也就是说添加或者inputs列表中的元素不会立刻中断select()方法.必须是新加的元素有改变的状态
.对outputs 列表监控两个内容 1个是列表中元素状态的改变 1个是列表中新加元素 ,也就是说outputs列表中新加元素也会立刻中断select()的阻塞,列表中的元素改变同样会中断.删除元素不会中断
.exceptional 这个还不知道,可能和 outputs一样吧.我觉得理解上面两个即可.
那么怎么测试的呢? inputs列表的测试,只需要进行连接,不进行数据发送即可.如果新增也会触发中断,那么account= 的时候连接上了,紧接着 就会中断account = 的select(),因为inputs列表新加了连接.这时代码会走data=s.recv()在继续就是删除这个连接了.但实际情况不是,程序代码会阻塞在accout=2的那次循环的select()处.
outputs 列表的测试,客户端连接后在send一个数据.服务端会建立连接后,把数据加入到outputs.append(s)中,此时account = 的select()方法就会中断阻塞,立马发送给客户端的操作,发送后相当于outputs列表中的元素执行了s.send()方法,所以会在account = 时再次中断select()方法,然后显示队列为空,删除outputs列表中的S元素.
总之: 记住上面3个结论.
'''
# Handle inputs
for s in readable: # 当select()阻塞中断了,说明有元素发生变化.进而先检查 readable列表 if s is server: # 如果时server实例发生变化,说明有新的连接
# A "readable" server socket is ready to accept a connection
connection, client_address = s.accept() # 实例化新连接
print('new connection from', client_address)
connection.setblocking(False) # 将新连接设置为阻塞状态
inputs.append(connection) # Give the connection a queue for data we want to send
message_queues[connection] = queue.Queue() # 为新连接创建一个queue实例
else: # 如果readable列表中有其他对象,说明是已连接的实例
data = s.recv() # 已连接的链接实例处于readable列表中说明有接收数据的请求.
if data: # 如果接收数据不为空
# A readable client socket has data
print(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()) )
message_queues[s].put(data) # 把接收到的数据,放到自己的queue队列中
# Add output channel for response
if s not in outputs: # 把此链接实例加入到outputs列表 ,下次循环到select()时加入到writable
outputs.append(s) #
else:
# Interpret empty result as closed connection # 如果数据为空,说明客户端关闭链接了
print('closing', client_address, 'after reading no data')
# Stop listening for input on the connection
if s in outputs: # 客户端既然断开了,就把发送的列表也给删除
outputs.remove(s) #既然客户端都断开了，我就不用再给它返回数据了，所以这时候如果这个客户端的连接对象还在outputs列表中，就把它删掉
inputs.remove(s) #inputs中也删除掉,
s.close() #把这个连接关闭掉 # Remove message queue
del message_queues[s] # 把这个链接的消息队列也删掉
# Handle outputs
for s in writable: # 如果writable列表中有元素,说明有信息要发送
try:
next_msg = message_queues[s].get_nowait() # 得到这个链接实例的专用queue队列
except queue.Empty: # 如果没有获得信息,则把此链接元素从outputs列表中删除
# No messages waiting so stop checking for writability.
print('output queue for', s.getpeername(), 'is empty')
outputs.remove(s)
else:
print( 'sending "%s" to %s' % (next_msg, s.getpeername())) # 如果得到信息那么发送
s.send(next_msg)
# Handle "exceptional conditions"
for s in exceptional: # 这个就过吧
print('handling exceptional condition for', s.getpeername() )
# Stop listening for input on the connection
inputs.remove(s)
if s in outputs:
outputs.remove(s)
s.close() # Remove message queue
del message_queues[s]
account +=
select_socket_server
client端：
#!/usr/bin/env python3.
#__author__:'ted.zhou'
''' '''
__author__ = 'jieli'
import socket
import sys
import time # messages = [ 'This is the message. ',
# 'It will be sent ',
# 'in parts.',
# ]
messages = ['']
server_address = ('localhost', ) # Create a TCP/IP socket
# socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
# socket.socket(socket.AF_INET, socket.SOCK_STREAM),
# ]
socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),] # Connect the socket to the port where the server is listening
print(sys.stderr, 'connecting to %s port %s' % server_address)
for s in socks:
s.connect(server_address) time.sleep() for message in messages: # Send messages on both sockets
for s in socks:
print(sys.stderr, '%s: sending "%s"' % (s.getsockname(), message))
s.send(bytes(message,'utf8')) # Read responses on both sockets
for s in socks:
data = s.recv()
print(sys.stderr, '%s: received "%s"' % (s.getsockname(), data))
if not data:
print(sys.stderr, 'closing socket', s.getsockname())
s.close()
socket_client
测试结果，可参考：
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> starting up on localhost port waiting for the next event
new connection from ('127.0.0.1', )waiting for the next event
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'11'" from ('127.0.0.1', )waiting for the next event
sending "b'11'" to ('127.0.0.1', )waiting for the next event
output queue for ('127.0.0.1', ) is emptywaiting for the next event
closing ('127.0.0.1', ) after reading no datawaiting for the next event
测试结果
