Python使用异步线程池实现异步TCP服务器交互

背景：

实现客户端与服务端交互，由于效率原因，要发送与接收异步，提高效率。

需要多线程，本文用线程池管理。

common代码：

import pickle
import struct
import timedef send_msg(conn, data):time.sleep(1)msg = pickle.dumps(data)msg = struct.pack('>I', len(msg)) + msgconn.sendall(msg)return data, len(msg)def recv_from(conn, n):data = b''handle_len = 0while handle_len < n:packet = conn.recv(n - handle_len)if not packet:return Nonehandle_len += len(packet)data += packetreturn datadef recv_msg(conn):struct_msg_len = recv_from(conn, 4)if not struct_msg_len:return None, 0msg_len = struct.unpack('>I', struct_msg_len)[0]msg = recv_from(conn, msg_len)msg = pickle.loads(msg)return msg, msg_len

客户端：

#!/usr/bin/python
# -*- coding: utf-8 -*-import time
import select
import socket
import threading
from threading import Thread
from concurrent.futures import as_completed
from concurrent.futures import ThreadPoolExecutorfrom common import send_msg, recv_msgsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 生成socket
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 不经过WAIT_TIME，直接关闭
sock.setblocking(False)  # 设置非阻塞编程inputs = [sock, ]
executor = ThreadPoolExecutor(max_workers=3)  # 设置线程池最大数量print('client start!!!')try:sock.connect(("127.0.0.1", 789))
except Exception as e:print(e)def handle_received_data(data):print("接收服务端信息:", data)time.sleep(1)returndef receive_service_data():"""接收服务端返回的数据并处理"""while True:try:r_list, w_list, e_list = select.select(inputs, [], [], 1)for event in r_list:data, data_len = recv_msg(event)if data:try:executor.submit(handle_received_data, data)except Exception as e:print(threading.current_thread(), threading.active_count())print(e)else:print("远程断开连接")inputs.remove(event)exit()except OSError as e:import tracebackprint(traceback.format_exc())print(e)exit()def send_client_data(size=100):"""发送客户端数据"""executors = []for i in range(size):exe = executor.submit(send_msg, sock, {'data': i})executors.append(exe)for feature in as_completed(executors):try:data, data_len = feature.result()except Exception as e:print(e)else:print(f"客户端发送数据：{data}, len:{data_len}")if __name__ == '__main__':T1 = time.time()# 启动接受服务端数据的线程Thread(target=receive_service_data).start()# 发送客户端数据send_client_data(size=10)print('all_time:', time.time() - T1)

服务端：

#!/usr/bin/python
# -*- coding: utf-8 -*-
import time
import socket
import select
import threading
from concurrent.futures import ThreadPoolExecutorfrom common import send_msg, recv_msgsock = socket.socket()
sock.bind(('127.0.0.1', 789))
sock.setblocking(False)
sock.listen()inputs = [sock, ]
lock = threading.Lock()
executor = ThreadPoolExecutor(max_workers=3)  # 设置线程池最大数量print('service start!!!')def handle_received_data(event, data):time.sleep(1)send_msg(event, data)print(f"服务端发送数据：{data}")while True:r_list, w_list, e_list = select.select(inputs, [], [], 1)for event in r_list:if event == sock:print("新的客户端连接")new_sock, addresses = event.accept()inputs.append(new_sock)else:data, msg_len = recv_msg(event)if data:print("接收到客户端信息", data)executor.submit(handle_received_data, event, data)else:print("客户端断开连接")inputs.remove(event)

运行结果：

---

参考：

Python select.select 模块通信全过程详解_南淮北安的博客-CSDN博客 

Python标准库socketserver使用线程混入实现异步TCP服务器 

Python中的多路复用 (select、poll 和 epoll) 

 Python实现socket的非阻塞式编程 - 简书

https://www.cnblogs.com/i-honey/p/8078518.html 

 Python多线程RuntimeError: can’t start new thread-Grugsum's blog

python自学成才之路 线程间协作之Semaphore，threading.local() - 腾讯云开发者社区-腾讯云 

Python can‘t start new thread_零之领域的博客-CSDN博客 

python socket sendto函数返回值_他拍了拍你，来这里看这个Socket。_weixin_39982580的博客-CSDN博客 python线程数设置多少_为什么线程数增加超过threading.BoundedSemaphore在python的设置？..._weixin_39808803的博客-CSDN博客

https://www.cnblogs.com/shuopython/p/14943175.html 

Python线程池及其原理和使用（超级详细）