Python select IO多路复用

1、select介绍

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

 

2、使用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中都删除，再把链接关闭掉

3、完整代码

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))