一、什么是 websocket？

websocket 是一种在客户端和服务器之间建立持久性连接的通信协议。它允许服务器主动向客户端推送数据，而不需要客户端发起请求。相比传统的 HTTP 请求 - 响应模式，Websocket 提供了更低的延迟和更高的实时性。

二、websocket 的测试指标

1. 建立连接时间：
连接建立时间是指从客户端发起链接请求到与服务器成功建立连接所花费的时间。对于实时性要求较高的应用，较短的连接建立时间是一个重要指标。

2. 数据传输延迟：
数据传输延迟是指从服务器发送数据到客户端接收数据之间所经过的时间。较低的延迟可以提供更好的实时性和用户体验。

3. 吞吐量：
吞吐量是指单位时间内传输的数据量。对于搞并发场景下的应用，较高的吞吐量可以提供更好的性能。

4. 客户端连接数：
客户端连接数表示同时与服务器建立连接的客户端数量。对于支持大规模并发访问的应用，能够同时处理大量客户端连接是一个重要的指标。

5. 连接稳定性：
连接稳定性是指在长时间运行的情况下，Websocket 连接是否能保持稳定.需要测试在高负载、长时间运行等场景下，连接是否会出现异常断开或延迟增加的情况。

三、websocket 测试方法

1. 连接时间测试

import time
import asyncio
import websockets
from config.websocketConfig import websocket_url


async def connection_time(uri, num_clients):
    try:
        clients = []
        times = 0

        async with websockets.connect(uri) as websocket:
            start_time = time.time()
            for _ in range(num_clients):

                client = await websockets.connect(uri)
                end_time = time.time()
                connection_time = end_time - start_time
                clients.append((client, connection_time))

            print(f" {num_clients} 个客户端成功连接至客websocket服务器.")

            for client, connection_time in clients:
                print(f"客户端 {client} 连接花费 {connection_time:.2f} 秒.")
                times = times + connection_time

            # 计算平均连接时间
            avg_time = times / num_clients
            print(f" 平均连接时间： {avg_time}秒")

    except Exception as e:
        print(f"Connection failed. Error message: {str(e)}")


async def main():
    # WebSocket 服务器地址
    uri = websocket_url
    # 客户端数量
    num_clients = 1000

    await connection_time(uri, num_clients)


if __name__ == '__main__':
    asyncio.run(main())

2. 数据传输延迟测试

from config.websocketConfig import websocket_url
import time
import asyncio
import websockets
from datetime import datetime


async def transfer_latency(uri, num_clients, message_size):
    try:
        clients = []
        async with websockets.connect(uri) as websocket:
            for _ in range(num_clients):
                client = await websockets.connect(uri)
                clients.append(websocket)

            print(f" {num_clients} 个客户端成功连接至websocket服务器.")

            start_time = time.time()

            for websocket in clients:
                # 发送指定大小的数据
                await websocket.send("x" * message_size)

            for websocket in clients:
                # 接收服务器响应数据
                response = await websocket.recv()

            end_time = time.time()

            # 计算传输时间
            transfer_time = end_time - start_time
            # 计算传输延迟
            latency = transfer_time / num_clients

            print(f"传输时间为 {transfer_time:.2f} 秒.")
            print(f"平均传输时间为: {latency:.2f} 秒.")

    except Exception as e:
        print(f"错误，错误信息: {str(e)}")


async def main():
    # WebSocket 服务器地址
    uri = websocket_url
    # 客户端数量
    num_clients = 100
    # 数据大小（字节）
    message_size = 1024

    # 开始时间
    start_time = time.time()
    start_time1 = datetime.fromtimestamp(start_time)
    startTime = start_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("开始时间：", startTime)

    await transfer_latency(uri, num_clients, message_size)

    # 结束
    end_time = time.time()
    end_time1 = datetime.fromtimestamp(end_time)
    endTime = end_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("结束时间：", endTime)


if __name__ == '__main__':
    asyncio.run(main())

3. 吞吐率测试

import asyncio
import websockets
import time
from config.websocketConfig import websocket_url
from datetime import datetime


async def send_and_receive_messages(uri, num_clients, num_messages, message_size):
    async def client_handler():
        async with websockets.connect(uri) as websocket:
            for _ in range(num_messages):
                await websocket.send("x" * message_size)
                await websocket.recv()

    tasks = [client_handler() for _ in range(num_clients)]
    await asyncio.gather(*tasks)


async def main():
    # WebSocket 服务器地址
    uri = websocket_url
    # 客户端数量
    num_clients = 1
    # 每个客户端发送的消息数量
    num_messages = 1
    # 数据大小
    message_size = 100

    # 开始时间
    start_time = time.time()
    start_time1 = datetime.fromtimestamp(start_time)
    startTime = start_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("开始时间：", startTime)
    await send_and_receive_messages(uri, num_clients, num_messages, message_size)
    # 结束
    end_time = time.time()
    end_time1 = datetime.fromtimestamp(end_time)
    endTime = end_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("结束时间：", endTime)

    # 总消息数量
    total_messages = num_clients * num_messages
    # 总耗时
    elapsed_time = end_time - start_time
    # 吞吐率 = 总消息数*单个信息大小/总耗时
    throughput = total_messages * message_size / elapsed_time
    print(
        f"总信息数: {total_messages}, 总共花费时间: {elapsed_time} 秒, 吞吐量: {throughput:.2f} b/s")


if __name__ == '__main__':
    asyncio.run(main())

4. 客户端连接瓶颈测试

import asyncio
import websockets
from config.websocketConfig import websocket_url
import time
from datetime import datetime


async def connectionNum(uri, num_clients):
    try:
        clients = []
        for _ in range(num_clients):
            client = websockets.connect(uri)
            clients.append(client)

        await asyncio.gather(*clients)
        print(f" {num_clients} 个客户端成功连接至websocket服务器.")
    except Exception as e:
        print(f" {num_clients} 个客户端连接失败. 错误信息: {str(e)}")


async def main():
    uri = websocket_url  # WebSocket 服务器地址
    max_clients = 1000  # 最大客户端数量

    # 开始时间
    start_time = time.time()
    start_time1 = datetime.fromtimestamp(start_time)
    startTime = start_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("开始时间：", startTime)

    for num_clients in range(1, max_clients + 1):
        await connectionNum(uri, num_clients)

    # 结束
    end_time = time.time()
    end_time1 = datetime.fromtimestamp(end_time)
    endTime = end_time1.strftime('%Y-%m-%d %H:%M:%S')
    print("结束时间：", endTime)


if __name__ == '__main__':
    asyncio.run(main())

四、总结

以上是使用 python 编写的 websocket 性能测试脚本，可以直接根据需求修改参数进行使用。但需要注意的是，websocket 的性能受多方面影响，比如网络带宽、服务器性能、客户端设备性能等，实际使用时需要根据情况而做调整或多次测试。另外，测试的时候，可以搭配一些监控工具，如 promethues+grafana 等实时检测 websocket 服务器的主机情况。