Python使用異步線程池如何實(shí)現(xiàn)異步TCP服務(wù)器交互

更新時間：2023年11月10日 08:50:23 作者：Rnan-prince

這篇文章主要介紹了Python使用異步線程池如何實(shí)現(xiàn)異步TCP服務(wù)器交互問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

背景

實(shí)現(xiàn)客戶端與服務(wù)端交互，由于效率原因，要發(fā)送與接收異步，提高效率。

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

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)  # 不經(jīng)過WAIT_TIME，直接關(guān)閉
sock.setblocking(False)  # 設(shè)置非阻塞編程
 
inputs = [sock, ]
executor = ThreadPoolExecutor(max_workers=3)  # 設(shè)置線程池最大數(shù)量
 
print('client start!!!')
 
try:
    sock.connect(("127.0.0.1", 789))
except Exception as e:
    print(e)
 
 
def handle_received_data(data):
    print("接收服務(wù)端信息:", data)
    time.sleep(1)
    return
 
 
def receive_service_data():
    """接收服務(wù)端返回的數(shù)據(jù)并處理"""
    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("遠(yuǎn)程斷開連接")
                    inputs.remove(event)
                    exit()
        except OSError as e:
            import traceback
            print(traceback.format_exc())
            print(e)
            exit()
 
 
def send_client_data(size=100):
    """發(fā)送客戶端數(shù)據(jù)"""
    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"客戶端發(fā)送數(shù)據(jù)：{data}, len:{data_len}")
 
 
if __name__ == '__main__':
    T1 = time.time()
 
    # 啟動接受服務(wù)端數(shù)據(jù)的線程
    Thread(target=receive_service_data).start()
 
    # 發(fā)送客戶端數(shù)據(jù)
    send_client_data(size=10)
    print('all_time:', time.time() - T1)

服務(wù)端

#!/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)  # 設(shè)置線程池最大數(shù)量
 
print('service start!!!')
 
 
def handle_received_data(event, data):
    time.sleep(1)
    send_msg(event, data)
    print(f"服務(wù)端發(fā)送數(shù)據(jù)：{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)