python实现并发服务器实现方式(多线程/多进程/select/epoll)

python实现并发服务器实现方式(多线程/多进程/select/epoll)

 

并发服务器开发

并发服务器开发，使得一个服务器可以近乎同一时刻为多个客户端提供服务。实现并发的方式有多种，下面以多进程，多线程，IO多路复用等方式实现并发。这里使用网络编程中的TCP服务器和客户端通信为例子。

多进程并发阻塞

利用进程把客户端和服务器进行管理，当有新的客户端连接到服务器时，就创建一个新的进程来管理，通过操作系统的调度，从而实现了并发的操作

from multiprocessing import Process
from socket import *


def recv_data(new_socket, client_info):
    print("客户端{}已经连接".format(client_info))
    # 接受数据
    raw_data = new_socket.recv(1024)
    while raw_data:
        print(f"收到来自{client_info}的数据：{raw_data}")
        raw_data = new_socket.recv(1024)
    new_socket.close()


def main():
    # 实例化socket对象
    socket_server = socket(AF_INET, SOCK_STREAM)
    # 设置端口复用
    socket_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    # 绑定IP地址和端口
    socket_server.bind(("", 7788))
    # 改主动为被动，监听客户端
    socket_server.listen(5)
    while True:
        # 等待连接
        new_socket, client_info = socket_server.accept()
        p = Process(target=recv_data, args=(new_socket, client_info))
        p.start()
        # 多进程会复制父进程的内存空间，所以父进程中new_socket也必须关闭
        new_socket.close()


if __name__ == '__main__':
    main()

多线程并发阻塞

多线程和多进程类似，只是线程间共享内存空间，要注意变量的管理

from threading import Thread
from socket import *


def recv_data(new_socket, client_info):
    print("客户端{}已经连接".format(client_info))
    # 接受数据
    raw_data = new_socket.recv(1024)
    while raw_data:
        print(f"收到来自{client_info}的数据：{raw_data}")
        raw_data = new_socket.recv(1024)
    new_socket.close()


def main():
    # 实例化socket对象
    socket_server = socket(AF_INET, SOCK_STREAM)
    # 设置端口复用
    socket_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    # 绑定IP地址和端口
    socket_server.bind(("", 7788))
    # 改主动为被动，监听客户端
    socket_server.listen(5)
    while True:
        # 等待连接
        new_socket, client_info = socket_server.accept()
        p = Thread(target=recv_data, args=(new_socket, client_info))
        p.start()
        # 多线程共享一片内存区域，所以这里不用关闭
        # new_socket.close()


if __name__ == '__main__':
    main()

多路复用IO---select模型

在操作系统层面上，系统提供了一个select接口，它会轮询给定的文件描述符状态，如果其中有描述符的状态改变，select()就会返回有变化的文件描述符。

from socket import *
import select


# 实例化对象
socket_server = socket(AF_INET, SOCK_STREAM)
# 绑定IP和端口
socket_server.bind(("", 7788))
# 将主动模式改为被动模式
socket_server.listen(5)
# 创建套接字列表
socket_lists = [socket_server]
# 等待客户端连接
while True:
    # 只监听读的状态，程序阻塞在这，不消耗CPU，如果列表里面的值读状态变化后，就解阻塞
    read_lists, _, _ = select.select(socket_lists, [], [])
    # 循环有变化的套接字
    for sock in read_lists:
        # 判断是否是主套接字
        if sock == socket_server:
            # 获取新连接
            new_socket, client_info = socket_server.accept()
            print(f"客户端：{client_info}已连接")
            # 添加到监听列表中
            socket_lists.append(new_socket)
        else:
            # 不是主客户端，即接收消息
            raw_data = sock.recv(1024)
            if raw_data:
                print(f"接收数据：{raw_data.decode('gb2312')}")
            else:
                # 如果没有数据，则客户端断开连接
                sock.close()
                # 从监听列表中删除该套接字
                socket_lists.remove(sock)

优点：良好的跨平台支持

缺点：1.监测的文件描述符数量有最大限制，Linux系统一般为1024，可以修改宏定义或者内核进行修改，但是会造成效率低下；2.对文件描述符采用轮询机制，每个文件描述符都会询问一遍，这样很消耗CPU时间

多路复用IO---epoll模型

为了解决select轮询机制造成的效率低下问题，则引入了epoll接口。相较于select的两大优势。1.没有文件描述符最大数量的限制(最大数量则看内存大小)；2.采用时间通知机制，当文件描述符状态有变时，主动通知内核进行调度。其中print注释是为了打印对象，查看对象是什么。

from socket import *
import select


# 创建socket对象
sock_server = socket(AF_INET, SOCK_STREAM)
# 绑定IP和端口
sock_server.bind(("", 7788))
# 将主动模式设置为被动模式，监听连接
sock_server.listen(5)
# 创建epoll监测对象
epoll = select.epoll()
# print("未注册epoll对象：{}".format(epoll))
# 注册主套接字,监控读状态
epoll.register(sock_server.fileno(), select.EPOLLIN)
# print("注册了主套接字后：{}".format(epoll))
# 创建字典，保存套接字对象
sock_dicts = {}
# 创建字典，保存客户端信息
client_dicts = {}
while True:
    # print("所有套接字：{}".format(sock_dicts))
    # print("所有客户端信息：{}".format(client_dicts))
    # 程序阻塞在这，返回文件描述符有变化的对象
    poll_list = epoll.poll()
    # print("有变化的套接字：{}".format(poll_list))
    for sock_fileno, events in poll_list:
        # print("文件描述符：{}，事件：{}".format(sock_fileno, events))
        # 判断是否是主套接字
        if sock_fileno == sock_server.fileno():
            # 创建新套接字
            new_sock, client_info = sock_server.accept()
            print(f"客户端：{client_info}已连接")
            # 注册到epoll监测中
            epoll.register(new_sock.fileno(), select.EPOLLIN)
            # 添加到套接字字典当中
            sock_dicts[new_sock.fileno()] = new_sock
            client_dicts[new_sock.fileno()] = client_info
        else:
            # 接收消息
            raw_data = sock_dicts[sock_fileno].recv(1024)
            if raw_data:
                print(f"来自{client_dicts[sock_fileno]}的数据：{raw_data.decode('gb2312')}")
            else:
                # 关闭连接
                sock_dicts[sock_fileno].close()
                # 注销epoll监测对象
                epoll.unregister(sock_fileno)
                # 数据为空，则客户端断开连接，删除相关数据
                del sock_dicts[sock_fileno]
                del client_dicts[sock_fileno]

 

 

IO多路复用和线程池在提高并发性上应用场景的区别

多路复用适用于需要保持大量闲置（区别于计算密集型）长连接的业务场景，例如聊天室。这样的好处是能够避免不断的创建新线程，导致系统资源浪费。需要注意，多路复用本质上是复用单线程的，回调函数的执行必然是有可能长时间阻塞的，所以如果涉及到耗时的计算密集型任务，则会大大降低系统处理其它连接的响应速度。

线程池则适合短连接并发的情况，比如普通的web业务系统，Tomcat的Servlet容器默认选择就是线程池（虽然3.0后支持异步，但一般情况下不常使用）。由于处理短连接的线程很快会退出，因此能够充分发挥线程池复用线程的好处。

当然，多路复用和线程池可以结合起来使用，效果也许更好，但代码复杂度也会相应提高，需要更好的设计。建议根据业务场景选择相应的技术，避免过早优化。

 

一点补充：很多人不知道协程该归于哪个技术范畴。协程除了在用户态通过栈切换实现控制流的切换以外，还通常将多路复用和线程池结合起来。比如go语言内置的协程就是在多线程的基础上实现了一套调度策略，调度策略的实现建立在操作系统内核提供的IO多路复用技术之上，同时go语言参考计算机硬件情况自动将协程绑定在若干个系统线程之上，从而实现资源的高效率利用。