我有这两个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 脚本并将其用作测试。
最终编写了我自己的线程基准测试。
为您的代码进行 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 测试 您将转到面板,然后输入有多少用户以及每秒有多少用户,然后运行测试