Xiaozhi WebSocket 并发测试工具

WebSocket 并发测试工具，用于测试 Xiaozhi 服务的性能和稳定性。

📌 项目说明

本项目是为 Xiaozhi ESP32 Server Java 开发的并发测试工具。

关于 Xiaozhi ESP32 Server Java

Xiaozhi ESP32 Server Java 是基于 Xiaozhi ESP32 项目开发的 Java 版本服务端，包含完整的前后端管理平台。该项目旨在为用户提供一个功能丰富、操作便捷的管理界面，帮助用户更好地管理设备、配置等。

考虑到企业级应用场景的需求，Java 作为一种成熟的企业级开发语言，具备更完善的生态系统支持和更强大的并发处理能力，因此我们选择开发这个 Java 版本的服务端，为项目提供更多可能性和扩展空间。

技术栈：

后端框架：Spring Boot + Spring MVC
前端框架：Vue.js + Ant Design
数据存储：MySQL + Redis
全局响应式：适配各种设备及分辨率

项目地址： https://github.com/joey-zhou/xiaozhi-esp32-server-java

✨ 特性

🚀 高并发测试：支持高并发 WebSocket 连接
🔄 多轮测试：支持每个客户端执行多轮测试，评估持续负载表现
📊 详细指标：连接时延、语音识别时延、音频响应时延等全方位指标
📈 实时监控：实时显示进度条和测试状态
📉 可视化报告：自动生成性能分布图表
🔍 详细日志：分级日志系统，便于调试和问题追踪
🎯 模块化设计：清晰的代码结构，易于维护和扩展

📁 项目结构

xiaozhi-concurrent/
├── src/                      # 源代码目录
│   ├── __init__.py          # 包初始化
│   ├── config.py            # 配置和常量
│   ├── logger.py            # 日志配置
│   ├── progress.py          # 进度跟踪
│   ├── metrics.py           # 测试指标收集
│   ├── client.py            # WebSocket 测试客户端
│   ├── tester.py            # 并发测试器
│   └── utils.py             # 工具函数
├── run_test.py              # 主入口文件
├── requirements.txt         # 依赖包列表
├── README.md                # 本文档
└── logs/                    # 日志和输出目录（自动创建）
    ├── xiaozhi_debug_*.log          # 调试日志
    ├── xiaozhi_client_flow_*.log    # 客户端流程日志
    └── xiaozhi_client_flow_sorted_*.log  # 按设备排序的流程日志

📦 安装

1. 克隆或下载项目

cd ~/documents
git clone https://github.com/joey-zhou/xiaozhi-concurrent.git
cd xiaozhi-concurrent

2. 安装依赖

方法 A：使用 pip

pip install -r requirements.txt

方法 B：使用 conda（推荐）

conda create -n test python=3.9
conda activate test
pip install -r requirements.txt

3. 依赖说明

必需依赖：

websockets - WebSocket 客户端
numpy - 数值计算
psutil - 系统监控

可选依赖：

matplotlib - 图表可视化（强烈推荐）
opuslib - Opus 音频编解码（如需真实音频测试）

🚀 快速开始

⚠️ 服务端准备（重要）

在运行测试之前，需要先配置服务端以支持测试设备的自动绑定。测试工具会生成以下格式的设备ID：

设备ID命名规则：xiaozhi-test-XXXXXX

例如：xiaozhi-test-000001、xiaozhi-test-000002 ... xiaozhi-test-000100
设备数量由 --clients 参数决定

由于服务端默认对未绑定的设备会生成验证码返回，您需要选择以下任一方案来处理测试设备：

方案一：修改服务端代码（推荐）

修改服务端实现自动绑定逻辑，使测试设备自动绑定到默认角色下。

文件路径: src/main/java/com/xiaozhi/communication/common/MessageHandler.java

方法: afterConnection()

代码行: 107-129

// 如果设备未绑定，尝试自动绑定到默认智能体
if (device.getRoleId() == null) {
    try {
        // 系统默认admin用户
        device.setUserId(1);
        device.setType("test");
        device.setDeviceName("test");
        int result = deviceService.add(device);
        if (result > 0) {
            logger.info("设备自动绑定到默认智能体成功 - DeviceId: {}", deviceId);
            // 重新查询设备信息以获取绑定后的完整信息
            SysDevice updatedDevice = deviceService.selectDeviceById(deviceId);
            if (updatedDevice != null) {
                device = updatedDevice;
            }
        } else {
            logger.warn("设备自动绑定失败，设备可能已存在 - DeviceId: {}", deviceId);
        }
    } catch (Exception e) {
        logger.error("设备自动绑定到默认智能体失败 - DeviceId: {}", deviceId, e);
    }
}

功能说明:

当设备首次连接且未配置角色时（roleId == null），自动创建设备记录
默认绑定到 userId=1 的用户（需根据实际业务调整）
设备类型和名称均设置为 "test"
绑定成功后重新查询完整的设备信息

方案二：数据库预插入设备记录

编写 SQL 脚本将测试设备预先插入到数据库中。设备ID需遵循 xiaozhi-test-XXXXXX 格式。

-- 示例：插入100个测试设备（根据实际需要调整数量）
-- 注意：roleId 需要根据实际数据库中存在的角色ID进行调整
INSERT INTO `xiaozhi`.`sys_device` (`deviceId`, `deviceName`, `roleId`, `type`, `userId`) VALUES
('xiaozhi-test-000001', 'test', 1, 'test', 1),
('xiaozhi-test-000002', 'test', 1, 'test', 1),
('xiaozhi-test-000003', 'test', 1, 'test', 1),
-- ... 更多设备
('xiaozhi-test-000100', 'test', 1, 'test', 1);

字段说明：

deviceId：设备ID（必需），格式为 xiaozhi-test-XXXXXX

deviceName：设备名称（必需），测试设备统一使用 "test"

roleId：角色ID（可选），需要对应数据库中已存在的角色

type：设备类型（可选），测试设备使用 "test"

userId：创建人ID（必需），默认为 1（admin用户）

其他字段（state、createTime 等）会使用数据库默认值

提示：方案一更加灵活，可以自动处理任意数量的测试设备，推荐用于开发和测试环境。

基础使用

# 使用默认参数运行测试（20个并发客户端）
conda activate test
python run_test.py

自定义参数

# 指定服务器地址和并发客户端数量
python run_test.py --server ws://192.168.1.100:8091/ws/xiaozhi/v1/ --clients 200

# 多轮测试（100个客户端，每个执行5轮）
python run_test.py --clients 100 --rounds 5

# 压力测试（200个客户端，每个执行10轮）
python run_test.py --clients 200 --rounds 10 --server ws://192.168.1.100:8091/ws/xiaozhi/v1/

# 使用自定义音频文件
python run_test.py --audio test_audio.wav --clients 50

查看帮助

python run_test.py --help

📊 输出说明

实时显示

测试运行时会显示：

进度条：测试完成进度和预计剩余时间
状态统计：各阶段客户端数量（连接中、Hello、唤醒词、音频发送、音频接收等）
性能指标：连接成功率、Hello成功率、唤醒词成功率和时延、音频帧率、真实对话响应等

日志文件

所有日志保存在 logs/ 目录：

xiaozhi_debug_[timestamp].log - 详细调试日志
- 每个客户端的详细操作记录
- 错误和异常信息
xiaozhi_client_flow_[timestamp].log - 客户端流程日志
- 按时间顺序的流程记录
- 精确到毫秒的时间戳
xiaozhi_client_flow_sorted_[timestamp].log - 按设备排序的流程日志
- 按设备ID分组，便于查看单个客户端的完整流程

性能报告

测试完成后会显示详细的性能统计：

+==========+===================+==========+
| 指标     | 成功率/计数       | 平均耗时 |
+==========+===================+==========+
| 连接     | 100.00% (100/100) | 0.154s   |
| Hello    | 100.00% (100/100) | 0.184s   |
| 唤醒词   | 100.00% (100/100) | 0.459s   |
| 音频接收 | 成功200, 超时0    | -        |
+----------+-------------------+----------+
+==============+=============================+
| 音频帧率统计 | 值                          |
+==============+=============================+
| 平均间隔     | 59.60ms                     |
| 质量         | 良好 (673/8321 延迟, 8.09%) |
| 范围         | 最小0.1ms / 最大762.0ms     |
+--------------+-----------------------------+
+==============+=============================+
| 音频流量统计 | 值                          |
+==============+=============================+
| 接收/发送/总 | 0.98MB / 0.85MB / 1.82MB    |
| 平均每客户端 | 接收0.00MB / 发送0.01MB     |
| 平均速率     | 接收10.6kbps / 发送69.5kbps |
| 峰值速率     | 接收15.2kbps / 发送69.6kbps |
+--------------+-----------------------------+
+================+======================================+
| 真实音频测试   | 值                                   |
+================+======================================+
| 发送成功率     | 100.00% (100 次)                     |
| 语音识别准确率 | 100.00% (100/100)                    |
| 发送音频时长   | 1.021s                               |
| 语音识别时延   | 均值0.952s / 最小0.926s / 最大0.995s |
| 服务器处理     | 均值0.816s / 最小0.543s / 最大1.463s |
| 用户感知       | 均值1.767s / 最小1.499s / 最大2.431s |
+----------------+--------------------------------------+

可视化图表

如果安装了 matplotlib，会自动生成：

xiaozhi_latency_dist_[timestamp].png - 延迟分布直方图

连接时延分布
Hello响应时延分布
唤醒词响应时延分布
用户感知时延分布

🔧 配置说明

配置文件：`src/config.py`

# WebSocket 配置
DEFAULT_SERVER_URL = "ws://localhost:8091/ws/xiaozhi/v1/"
DEFAULT_CLIENTS = 20
DEFAULT_CONCURRENCY = 20

# 音频配置
AUDIO_SAMPLE_RATE = 16000
AUDIO_CHANNELS = 1
AUDIO_FRAME_DURATION_MS = 60

可根据需要修改这些默认值。

💡 使用技巧

1. 压力测试建议

逐步增加并发数和轮数，观察系统表现
建议从小规模开始：--clients 10 --rounds 1
根据服务器性能调整：
- 小型测试：--clients 50 --rounds 3
- 中型测试：--clients 100 --rounds 5
- 大型测试：--clients 500 --rounds 10

2. 调试技巧

查看详细日志：

# 实时查看调试日志
tail -f logs/xiaozhi_debug_*.log

# 按客户端分析流程
grep "xiaozhi-test-000001" logs/xiaozhi_client_flow_sorted_*.log

# 查看失败的客户端
grep -i "失败\|超时\|error" logs/xiaozhi_debug_*.log

🐛 常见问题

Q1: matplotlib 中文显示问题

A: 安装中文字体

# Ubuntu/Debian
sudo apt-get install fonts-wqy-microhei

# macOS
# 系统已自带中文字体

📈 模块说明

client.py - WebSocket 客户端

单个客户端的完整测试流程：

建立 WebSocket 连接
发送 Hello 握手
发送唤醒词探测
发送音频数据
接收和统计响应

tester.py - 并发测试器

管理多个客户端的并发测试：

管理多客户端并发连接
收集汇总指标
生成测试报告
实时显示进度

metrics.py - 指标收集

收集和统计各类性能指标：

连接时延
各阶段响应时延
成功率和失败率
音频流量统计

progress.py - 进度跟踪

测试进度和状态：

进度条显示
各阶段统计
完成/失败计数

🤝 贡献

欢迎提交 Issue 和 Pull Request！

Name		Name	Last commit message	Last commit date
Latest commit History 4 Commits
src		src
.gitignore		.gitignore
1加一等于几.wav		1加一等于几.wav
LICENSE		LICENSE
README.md		README.md
requirements.txt		requirements.txt
run_test.py		run_test.py

License

joey-zhou/xiaozhi-concurrent

Folders and files

Latest commit

History

Repository files navigation