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

 更新时间:2023年11月10日 08:50:23   作者:Rnan-prince  
这篇文章主要介绍了Python使用异步线程池如何实现异步TCP服务器交互问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

背景

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

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

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)

运行结果

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

相关文章

  • matplotlib实现热成像图colorbar和极坐标图的方法

    matplotlib实现热成像图colorbar和极坐标图的方法

    今天小编就为大家分享一篇matplotlib实现热成像图colorbar和极坐标图的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2018-12-12
  • Python Pygame实战之趣味篮球游戏的实现

    Python Pygame实战之趣味篮球游戏的实现

    这篇文章主要为大家分享了一个基于Python和Pygame实现的一个趣味篮球游戏,文中的示例代码讲解详细,对我们学习Python有一定帮助,需要的可以参考一下
    2022-04-04
  • 详解Django中类视图使用装饰器的方式

    详解Django中类视图使用装饰器的方式

    这篇文章主要介绍了详解Django中类视图使用装饰器的方式,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-08-08
  • python client使用http post 到server端的代码

    python client使用http post 到server端的代码

    python client使用 http post 到server端的代码,供大家学习参考
    2013-02-02
  • 对python3新增的byte类型详解

    对python3新增的byte类型详解

    今天小编就为大家分享一篇对python3新增的byte类型详解,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2018-12-12
  • python实现LRU热点缓存及原理

    python实现LRU热点缓存及原理

    LRU算法根据数据的历史访问记录来进行淘汰数据,其核心思想是“如果数据最近被访问过,那么将来被访问的几率也更高”。 。这篇文章主要介绍了python实现LRU热点缓存,需要的朋友可以参考下
    2019-10-10
  • Python3.7+tkinter实现查询界面功能

    Python3.7+tkinter实现查询界面功能

    这篇文章主要介绍了Python3.7+tkinter实现查询界面功能,本文通过实例代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下
    2019-12-12
  • 用Python编写个解释器实现方法接受

    用Python编写个解释器实现方法接受

    计算机只能理解机器码。归根结底,编程语言只是一串文字,目的是为了让人类更容易编写他们想让计算机做的事情。真正的魔法是由编译器和解释器完成,它们弥合了两者之间的差距。解释器逐行读取代码并将其转换为机器码
    2023-01-01
  • 如何利用python给微信公众号发消息实例代码

    如何利用python给微信公众号发消息实例代码

    使用过微信公众号的小伙伴应该知道微信公众号有时候会给你推一些文章,当你选择它的某个功能时,它还会返回一些信息,下面这篇文章主要给大家介绍了关于如何利用python给微信公众号发消息的相关资料,需要的朋友可以参考下
    2022-03-03
  • python中定时器的高级使用方式详解

    python中定时器的高级使用方式详解

    在Python编程中,定时器是一种非常有用的工具,用于执行特定任务或函数,本文将介绍一些高级的定时器使用方式,感兴趣的小伙伴可以跟随小编一起学习一下
    2024-02-02

最新评论