多线程的 udp client server 测试demo

import socket
import json
import uuid
from concurrent.futures import ThreadPoolExecutor

# 服务端配置
HOST = '127.0.0.1'
PORT = 9999
MAX_BUFFER_SIZE = 4096
MAX_THREADS = 500  # 最大并发线程数

# 创建 UDP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind((HOST, PORT))
print(f"UDP 服务端启动，监听地址 {HOST}:{PORT}")

# 使用线程池处理请求
executor = ThreadPoolExecutor(max_workers=MAX_THREADS)

def handle_request(data, client_address):
    try:
        # 解析 JSON 请求
        request = json.loads(data.decode('utf-8'))
        search_id = request.get('search_id')
        message = request.get('message', '')
        print(f"[请求] 来自 {client_address}，search_id: {search_id}")

        # 构造响应
        response = {
            "status": "success",
            "search_id": search_id,
            "message": f"已收到请求: {message}"
        }

        json_response = json.dumps(response).encode('utf-8')
        server_socket.sendto(json_response, client_address)
        print(f"[响应] 发送至 {client_address}，search_id: {search_id}")
    except json.JSONDecodeError:
        print(f"[错误] 接收的不是有效的 JSON 数据，来自 {client_address}")
    except Exception as e:
        print(f"[异常] 处理请求时发生错误: {e}")

# 主循环接收请求
try:
    while True:
        data, client_address = server_socket.recvfrom(MAX_BUFFER_SIZE)
        executor.submit(handle_request, data, client_address)
except KeyboardInterrupt:
    print("\n服务端关闭。")
    server_socket.close()

import socket
import json
import uuid
import threading
import time
import random

# 客户端配置
SERVER_HOST = '127.0.0.1'
SERVER_PORT = 9999
CLIENT_THREADS = 500  # 客户端线程数
TIMEOUT = 5  # 超时时间（秒）
MAX_BUFFER_SIZE = 32000

def client_thread_func(thread_id):
    # 每个线程使用自己的 socket
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    client_socket.settimeout(TIMEOUT)

    # 生成随机 UUID 作为 search_id
    search_id = str(uuid.uuid4())
    request = {
        "search_id": search_id,
        "message": f"这是来自线程 {thread_id} 的请求"
    }
    json_request = json.dumps(request).encode("utf-8")
    t1 = time.time()
    print(f"[客户端线程 {thread_id}] 发送请求，search_id: {search_id}")
    client_socket.sendto(json_request, (SERVER_HOST, SERVER_PORT))

    try:
        # 等待响应
        response_data, server_address = client_socket.recvfrom(MAX_BUFFER_SIZE)
        response = json.loads(response_data)
        print(f"[客户端线程 {thread_id}] 收到响应: {response}")
        t2 = time.time()
        time_use_ms = round(1000*(t2-t1), 3)
        # 校验 search_id
        if 'search_id' not in response:
            print(f"[客户端线程 {thread_id}] 响应中缺少 search_id 字段！ time_use_ms: {time_use_ms}")
        elif response['search_id'] != search_id:
            print(f"[客户端线程 {thread_id}] 错误！search_id 不匹配。发送的: {search_id}，返回的: {response['search_id']}. time_use_ms: {time_use_ms}")
        else:
            print(f"[客户端线程 {thread_id}] search_id 校验通过。 time_use_ms: {time_use_ms}")
        
    except socket.timeout:
        print(f"[客户端线程 {thread_id}] 超时，未收到响应。")
    except json.JSONDecodeError:
        print(f"[客户端线程 {thread_id}] 接收到的响应不是有效的 JSON 数据。")
    except Exception as e:
        print(f"[客户端线程 {thread_id}] 发生异常: {e}")
    finally:
        client_socket.close()

# 启动多个客户端线程
threads = []
for i in range(CLIENT_THREADS):
    t = threading.Thread(target=client_thread_func, args=(i,))
    threads.append(t)
    t.start()
    # 随机延迟
    time.sleep(random.uniform(0.0001, 0.001))

# 等待所有线程完成
for t in threads:
    t.join()

print("所有客户端线程执行完毕。")