Python 负载测试套接字通信

Question

我有这两个Python脚本server.py和client.py，它们使用套接字连接，服务器看起来像：

while True:
    try:
        conn, addr = server.accept()
        handle(conn, addr)
    except KeyboardInterrupt:
        break

所以它一次只处理一个客户端，稍后我将添加 gevent，以便它可以同时处理客户端。

如何对其进行基准测试以确定使用 gevent 与不使用 gevent 时的每秒客户端请求数？

我找到了很多 HTTP 基础基准测试工具，但是 TCP 呢？理想情况下，我希望测试运行我的几个 client.py 脚本并将其用作测试。

Answer 1

最终编写了我自己的线程基准测试。

Answer 2

为您的代码进行 Locust 测试，用于测试任何内容的负载，编写 Locust 脚本后，您应该运行服务器，然后开始测试。 蝗虫测试会是这样的

from locust import User, task, between, events
import websocket
import time

class WebSocketClient:
    def __init__(self, host):
        self.host = host
        self.conn = websocket.WebSocket()

    def connect(self):
        self.conn.connect(self.host)

    def send(self, message):
        start_time = time.time()  # Start timing before sending
        self.conn.send(message)
        return start_time  # Return the start time to calculate the round trip time

    def receive(self):
        return self.conn.recv()

    def close(self):
        self.conn.close()

class WebSocketUser(User):
    abstract = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)

    def on_start(self):
        self.client.connect()

    def on_stop(self):
        self.client.close()

class ApiUser(WebSocketUser):
    wait_time = between(1, 2)

    @task
    def send_and_receive(self):
        start_time = self.client.send("Hello, WebSocket!")
        response = self.client.receive()
        total_time = time.time() - start_time  # Calculate the total round trip time
        # Fire a custom event to report the request time
        events.request.fire(
            request_type="WebSocket",
            name="Echo",
            response_time=int(total_time * 1000),  # Convert to milliseconds
            response_length=len(response),
            exception=None,
            context=None
        )
        print(f"Received: {response}")

@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    print("Locust is starting, environment setup complete.")

@events.test_stop.add_listener
def on_locust_quit(environment, **_kwargs):
    print("Locust is stopping.")

你可以像 bash -> locust -f test_file.py 一样运行 locust 测试 您将转到面板，然后输入有多少用户以及每秒有多少用户，然后运行测试