背景:
实现客户端与服务端交互,由于效率原因,要发送与接收异步,提高效率。
需要多线程,本文用线程池管理。
common代码:
import pickle
import struct
import time
def send_msg(conn, data):
time.sleep(1)
msg = pickle.dumps(data)
msg = struct.pack('>I', len(msg)) + msg
conn.sendall(msg)
return data, len(msg)
def recv_from(conn, n):
data = b''
handle_len = 0
while handle_len < n:
packet = conn.recv(n - handle_len)
if not packet:
return None
handle_len += len(packet)
data += packet
return data
def recv_msg(conn):
struct_msg_len = recv_from(conn, 4)
if not struct_msg_len:
return None, 0
msg_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 ThreadPoolExecutor
from common import send_msg, recv_msg
sock = 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)
return
def 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 traceback
print(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 ThreadPoolExecutor
from common import send_msg, recv_msg
sock = 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线程池及其原理和使用(超级详细)