select实现IO多路复用服务器

Posted 2020-11-10 与君同悦

select模块：

import select, sys, socket,queue
"""
多路复用IO
典型案例： select, poll, epoll
select代码实现
"""


#创建套接字
server = socket.socket()
#设置非阻塞
server.setblocking(0)
#要监听的地址和端口
server_addr = ("0.0.0.0", 10000)
print("starting up on %s port %s" % server_addr)
#绑定地址和端口
server.bind(server_addr)
server.listen()
inputs = [server, ]  #自己也要检测，因为本身也是个fd
outputs = []
message_queues = {}

while True:
    print("waiting for next event...", len(inputs))
    #如果没有任何fd就绪，程序就一直卡在这里
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    for r in readable:
        #如果server有反应，代表有新连接来了
        if r is server:
            #新连接来了，接受这个连接
            conn, client_addr = r.accept()
            print("new connection from", client_addr)
            conn.setblocking(0)
            #把客户端连接状态加入inputs列表
            inputs.append(conn)
            #为每个连接建立一个消息队列
            message_queues[conn] = queue.Queue()
        #接收数据
        else:
            data = r.recv(1024)
            if data:
                print("收到来自[%s]的数据:" % r.getpeername()[0], data)
                #收到的数据放到队列里，等待发送
                message_queues[r].put(data)
                if r not in outputs:
                    #为了不影响处理其他客户端连接，不立即返回数据给客户端
                    outputs.append(r)
            #没有接收到数据，说明客户端断开
            else:
                print("客户端断开了", r)
                if r in outputs:
                    #清理已经断开的连接
                    outputs.remove(r)
                inputs.remove(r)
                del message_queues[r]
    #处理需要发送数据的连接
    for w in writeable:
        try:
            send_data = message_queues[w].get_nowait()
            print("sending msg to [%s]" % w.getpeername()[0], send_data)
            w.send(send_data.upper())
        except queue.Empty:
            print("client [%s]" % w.getpeername()[0], "queue is empty...")
            outputs.remove(w)
    #处理异常的连接
    for e in exceptional:
        print("handling exception for ", e.getpeername()[0])
        inputs.remove(e)
        if e in outputs:
            outputs.remove(e)
        del message_queues[e]
        e.close()

　　selectors：

import selectors, socket
"""
select模块的封装版本
使用更简单 
"""

sel = selectors.DefaultSelector()

#
def accept(sock, mask):
    conn, addr = sock.accept()
    print(‘accepted‘, conn, ‘from‘, addr)
    conn.setblocking(0)
    sel.register(conn, selectors.EVENT_READ, read)


def read(conn, mask):
    data = conn.recv(1024)
    if data:
        print(‘echoing‘, repr(data), ‘to‘, conn)
        conn.send(data)
    else:
        print(‘closing‘, conn)
        sel.unregister(conn)
        conn.close()


server = socket.socket()
server.bind(("0.0.0.0", 10000))
server.listen()
server.setblocking(0)
sel.register(server, selectors.EVENT_READ, accept)

while True:
    #作用同select.secect(),所有客户端连接都在这里维护
    events = sel.select()
    for key, mask in events:
        #取到注册的回调函数callback = accept
        callback = key.data
        callback(key.fileobj, mask)

　　模拟多个客户端和服务器交互数据:

import socket, sys

messages = [
    b‘hello xiaobai‘,
    b‘this is test message‘,
    b‘do not reply‘
]

server_addr = ("localhost", 10000)

socks = [socket.socket() for i in range(600)]

print(‘connecting to %s port %s‘ % server_addr)
for client in socks:
    client.connect(server_addr)
    for m in messages:
        print(‘%s:sending "%s"‘ % (client.getsockname(), m))
        client.send(m)
        data = client.recv(1024)
        print(‘%s:received "%s"‘ % (client.getsockname(), data))
        if not data:
            print(sys.stderr, ‘closing socket‘, s.getsockname())