Python使用selectors库开发高性能服务器

今天药忘吃喽~ 2022-10-28 00:48 162阅读 0赞

模拟网络通信，单独使用socket库编写时，你可能会这么写

> 我在Linux上运行的，没在Windows上运行过

**socket\_server.py 服务端**

# -*- coding: UTF-8 -*-
    import socket
    
    
    # 创建socket对象
    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 设置IP地址复用
    serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 3)
    # 绑定地址和端口
    serversocket.bind(('127.0.0.1', 8888))
    # 开始监听，最大监听个数设置为10个
    serversocket.listen(10)
    while(True):
        print('等待连接中......')
        # 等待建立客户端连接
        client_obj, client_address = serversocket.accept()
        print('有连接过来了')
        # 客户端发来的信息
        client_data = client_obj.recv(1020)
        print(client_address, '客户端说：', client_data.decode())
        # 回复客户端
        client_obj.sendall(bytes('我收到了你的信息', encoding='utf-8'))
        client_obj.close()
    
    serversocket.close()

**socket\_client.py 客户端**

# -*- coding: UTF-8 -*-
    import socket
    
    
    while(True):
        # 创建客户端socket对象
        clientsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 客户端连接指定的服务端
        clientsocket.connect(('127.0.0.1', 8888))
        data = input('请输入内容：')
        clientsocket.sendall(bytes(data, encoding='utf-8'))
        server_data = clientsocket.recv(1024)
        print('服务端发来消息说:', str(server_data, encoding='utf-8'))
        clientsocket.close()

像我这么写的话是能够运行，但是多开几个客户端时，是**不能正常**多个客户端**同时**与服务端通信的，因为服务端的I/O是阻塞的，如果要实现与多个客户端通信的效果，可以使用多线程的方法写，假设你会这么写，客户端不变，服务端修改为如下

# -*- coding: UTF-8 -*-
    import socket
    import _thread
    
    
    # 创建socket对象
    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 设置IP地址复用
    serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 3)
    # 绑定地址和端口
    serversocket.bind(('127.0.0.1', 8888))
    # 开始监听，最大监听个数设置为10个
    serversocket.listen(10)
    
    
    def accept_read(client_obj):
        print('有连接过来了')
        # 客户端发来的信息
        client_data = client_obj.recv(1020)
        print(client_address, '客户端说：', client_data.decode())
        # 回复客户端
        client_obj.sendall(bytes('我收到了你的信息', encoding='utf-8'))
        client_obj.close()
    
    
    while(True):
        print('等待连接中......')
        # 等待建立客户端连接
        client_obj, client_address = serversocket.accept()
        _thread.start_new_thread(accept_read, (client_obj,))
    
    serversocket.close()

这样实现了我们的要求，但是，线程的开销是非常大的，如果有多少个客户端就有多少个线程，那么一台普通机器能开启的线程数量是有限的，所以我们不采用这种方法。

我们采用I/O多路复用的方式使得服务端能够在单线程的条件下，支持与多个客户端同时通信。I/O多路复用技术有很多种，`select`、`poll`、`epoll`，epoll是select和poll的增强版，但不是任何情况下都是epoll最好，在并发量相对较小的情况下select和poll的效率较高，当并发量较大时epoll的效率最高，像redis和nginx都是使用的epoll机制，所以它们的并发量都很高。

python第三方库select库能实现以上三种机制，selectors库是select库的升级版，推荐使用selectors库，官方文档[https://docs.python.org/zh-cn/3.7/library/selectors.html][https_docs.python.org_zh-cn_3.7_library_selectors.html]

再次修改服务端的代码，如下，修改的地方不多，只是解释写的多一点

# -*- coding: UTF-8 -*-
    import socket
    import selectors
    
    
    def read_data(client, mask):
        data = client.recv(1024)  # 取出客户端发来的信息，该消息存放在缓冲区
        if data:  # 检查data变量是否有值，如果有
            print('客户端说：', data.decode())
            client.sendall(b'wellcom to leizhou!!!')  # 回复消息
        else:  # 如果没有
            ''' 由于前面没有注销对象，该read_data()函数会被多次执行 第一次取出缓冲区的数据后，第二次执行时就没有了，data为空 由于客户端每次发送消息都会进行与服务端的连接和关闭，所以ctrl+c退出客户端时，该方法还会执行一次 '''
            print('一个客户端关闭了连接')
            selectors_io.unregister(client)  # 注销本次客户端连接对象
            client.close()  # 关闭该客户端对象的连接。先注销再关闭
    
    
    def accept(sock, mask):
        client_obj, address = sock.accept()  # 创建客户端连接对象
        print('有个客户端连接过来了', address)
        client_obj.setblocking(False)  # 设置该连接为非阻塞
        selectors_io.register(client_obj, selectors.EVENT_READ, read_data)  # 注册文件对象，客户端连接对象接收到数据时就调用read_data方法
    
    
    if __name__ == '__main__':
        socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 创建socket对象
        socket_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 3)  # 设置IP地址复用
        socket_server.bind(('127.0.0.1', 8888))  # 绑定地址和端口
        socket_server.listen(10)  # 开始监听，最大监听个数设置为10个
        socket_server.setblocking(False)  # 设置为非阻塞
        selectors_io = selectors.DefaultSelector()  # 自动检查系统支持的I/O复用，并使用最高效率的I/O复用机制
        selectors_io.register(socket_server, selectors.EVENT_READ, accept)  # 注册文件对象，并监视其I/O事件，指明对读事件感兴趣，触发后执行accept函数
        while True:
            try:
                events_list = selectors_io.select()  # 等待某些注册的文件对象就绪，可设置超时时间
                # 返回一个(键、事件)元组列表，每个元组对应一个准备好的文件对象
                # key是准备好的文件对象对应的selector实例
                # 事件是这个文件对象上准备好的事件的位掩码
                for key, mask in events_list:
                    key.data(key.fileobj, mask)
                    # key.fileobj连接时为<socket.socket fd=3, family=AddressFamily.AF_INET, type=2049, proto=0, laddr=('127.0.0.1', 8888)>
                    # key.fileobj断开时为<socket.socket [closed] fd=-1, family=AddressFamily.AF_INET, type=2049, proto=0>
            except KeyboardInterrupt:  # ctrl+c关闭运行时都会引发该异常，强迫症不想看到异常
                print('=========关闭服务器=========\n')
                exit()
        selectors_io.close()  # 调用它以确保释放任何基础资源

[https_docs.python.org_zh-cn_3.7_library_selectors.html]: https://docs.python.org/zh-cn/3.7/library/selectors.html