Python select IO多路复用

一、select介绍

Python的select()函数是底层操作系统实现的直接接口。它监视套接字,打开文件和管道(任何带有返回有效文件描述符的fileno()方法),直到它们变得可读或可写,或者发生通信错误。select()使得更容易同时监视多个连接,并且比使用套接字超时在Python中编写轮询循环更有效,因为监视发生在操作系统网络层而不是解释器。

二、使用select编写SocketServer

接下来通过socket server例子要以了解select 是如何通过单进程实现同时处理多个非阻塞的socket连接的

首先我们建立一个服务器端socket

import select
import socket
import sys
from queue import Queue
import queue # 创建套接字
server = socket.socket()
server.setblocking(False)
server.bind(('localhost',8800))
# 监听传入的连接
server.listen(5)

select()的参数是三个包含要监视的通信通道的列表。第一个是要检查要读取的传入数据的对象列表,第二个包含在缓冲区中有空间时将接收传出数据的对象,第三个包含可能有错误的对象(通常是错误的组合输入和输出通道对象)。服务器的下一步是设置包含要传递给select()输入源和输出目标的列表。

input = [server]
output = []
select.select(input,output,input)

所有客户端的进来的连接和数据将会被server的主循环程序放在上面的list中处理,我们现在的server端需要等待连接可写(writable)之后才能过来,然后接收数据并返回(因此不是在接收到数据之后就立刻返回),因为每个连接要把输入或输出的数据先缓存到queue里,然后再由select取出来再发出去。

# 对外发送数据的队列,记录到字典中
message_queues = {}

服务器程序的主要部分循环,调用select()来阻止并等待网络活动。

while 1:
#等待至少一个套接字准备好处理
print(sys.stderr, '\nwaiting for the next event')
"""
readable中的socket连接代表有数据可接受,可读取
writable list中存放着你可以对其进行发送(send)操作
exceptional 当连接通信出现error时会把error写到exceptional列表中
"""
readable, writable, exceptional = select.select(input, output, input)
readable列表中的socket可能会有3种可能状态,
1. 如果这个socket是server(它负责监听客户端的连接),那就表示server端已经收到一个新的连接
2. 客户端把数据发送过来了
3. 客户端和服务器端断开了连接,将客户端对象从 列表和队列中删除
    for s in readable:
if s is server: # 第1种情况,表示有新的连接进来
connection,add = s.accept() # 接受新的连接
connection.setblocking(0) """
这个时候,为了不阻塞整个程序的运行,我们先将它放入input列表中。
下一次loop时,就会被select去监听,如果这个连接的客户端发来了数据
那么这个连接的fd在server端就会变成就绪的,select就会把这个连接返回到readable列表中
然后在 for s in readable中取出这个连接,开始接受数据
"""
input.append(connection)
message_queue[connection] = Queue() else: # 第2种情况就是,客户端把数据发送了过来
data = s.recv(1024) # 通过recv去接受数据
if data:
print(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()))
message_queue[s].put(data) # 接受到的数据先放到队列中
if s not in output:
output.append(s) # 为了不影响处理与其他客户端的连接,这里不立刻返回数据给客户端
else: # 第3种情况 就是客户端断开了连接,这个时候recv()数据就是空,这个时候就可以跟客户端断开连接
if s in output:
"""
既然断开了连接,就没有必要给客户端发送数据了
如果客户端连接对象还在output中,就把他删除
"""
output.remove(s)
input.remove(s) # 在input列表中也删除掉
# 关闭连接,在队列中也删除
s.close()
del message_queue[s]

对于writable list中的socket,也有几种状态,如果这个客户端连接在跟它对应的queue里有数据,就把这个数据取出来再发回给这个客户端,否则就把这个连接从output list中移除,这样下一次循环select()调用时检测到outputs list中没有这个连接,那就会认为这个连接还处于非活动状态

    for s in writable:
try:
next_msg = message_queue[s].get_nowait()
except queue.Empty as e:
output.remove(s)
else:
s.send(next_msg)

最后,exceptional 有异常产生。如果在跟某个socket连接通信过程中出了错误,就把这个连接对象在inputs\outputs\message_queue中都删除,再把连接关闭掉

三、完整代码

server端:

# 通过非阻塞io实现http请求
# select + 回调 + 事件循环 # 使用单线程完成高并发 import select
import socket
import sys
from queue import Queue
import queue # 创建套接字
server = socket.socket()
server.setblocking(False)
server.bind(('localhost',8800))
# 监听传入的连接
server.listen(5) """
第一个是要检查要读取的传入数据的对象列表,
第二个包含在缓冲区中有空间时将接收传出数据的对象,
第三个包含可能有错误的对象(通常是错误的组合输入和输出通道对象)。
"""
input = [server] # 从中读取数据
output = [] # 将数据发送出去
select.select(input,output,input) message_queue = {} # 消息队列 while input:
#等待至少一个套接字准备好处理
print(sys.stderr, '\nwaiting for the next event')
"""
readable中的socket连接代表有数据可接受,可读取
writable list中存放着你可以对其进行发送(send)操作
exceptional 当连接通信出现error时会把error写到exceptional列表中
"""
readable, writable, exceptional = select.select(input, output, input) """
readable列表中的socket可能会有3种可能状态,
1. 如果这个socket是server(它负责监听客户端的连接),那就表示server端已经收到一个新的连接
2. 客户端把数据发送过来了
3. 客户端和服务器端断开了连接,将客户端对象从 列表和队列中删除
"""
for s in readable:
if s is server: # 第一种情况,表示有新的连接进来
connection,add = s.accept() # 接受新的连接
connection.setblocking(0) """
这个时候,为了不阻塞整个程序的运行,我们先将它放入input列表中。
下一次loop时,就会被select去监听,如果这个连接的客户端发来了数据
那么这个连接的fd在server端就会变成就绪的,select就会把这个连接返回到readable列表中
然后在 for s in readable中取出这个连接,开始接受数据
"""
input.append(connection)
message_queue[connection] = Queue() else: # 第2种情况就是,客户端把数据发送了过来
data = s.recv(1024) # 通过recv去接受数据
if data:
print(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()))
message_queue[s].put(data) # 接受到的数据先放到队列中
if s not in output:
output.append(s) # 为了不影响处理与其他客户端的连接,这里不立刻返回数据给客户端
else: # 第3种情况 就是客户端断开了连接,这个时候recv()数据就是空,这个时候就可以跟客户端断开连接
if s in output:
"""
既然断开了连接,就没有必要给客户端发送数据了
如果客户端连接对象还在output中,就把他删除
"""
output.remove(s)
input.remove(s) # 在input列表中也删除掉
# 关闭连接,在队列中也删除
s.close()
del message_queue[s] """
writable list也有几种状态,如果客户端连接在跟它对应的queue里有数据时,就把这个数据取出来再发给用户
否则就把这个连接从output中移除,这样下一次,select调用时检测到output列表中没有这个连接,就会认为这个连接处于非活动状态
"""
for s in writable:
try:
next_msg = message_queue[s].get_nowait()
except queue.Empty as e:
output.remove(s)
else:
s.send(next_msg) """
如果跟某个socket连接通信失败出现错误,就把这个连接对象从 各个列表中删除,再关闭连接
"""
for s in exceptional:
input.remove(s)
for s in output:
output.remove(s)
s.close()
del message_queue[s]

client端:

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('localhost',8800)) while True:
msg = bytes(input("<<<"),encoding='utf-8')
client.sendall(msg) data = client.recv(1024) print("{}".format(data))
上一篇:积累一下SQL


下一篇:用最基本的EF+MVC+JQ+AJAX+bootstrap实现权限管理的简单实例 之登陆和操作权限