当前位置：首页 > news >正文

WebSocket压力测试实战：连接保活、吞吐量与并发连接数深度解析

news 2026/7/28 12:34:33

1. 项目概述：为什么WebSocket压力测试是后端服务的“体检报告”

在实时交互应用大行其道的今天，无论是金融交易行情推送、在线协同文档编辑，还是多人在线游戏、即时通讯，其背后都离不开一个核心协议：WebSocket。与传统的HTTP请求-响应模式不同，WebSocket建立的是全双工、长连接通道，数据可以随时在客户端与服务器之间双向流动。这带来了极佳的实时体验，但也对后端服务的稳定性、资源管理能力和性能边界提出了前所未有的挑战。

很多开发者容易陷入一个误区：用HTTP接口的压力测试经验直接套用到WebSocket上。结果就是，上线后才发现服务在几百个长连接下就内存泄漏、在消息洪峰时响应延迟飙升、或者连接莫名其妙地大面积断开。这些问题的根源在于，WebSocket的压力测试维度与传统HTTP有本质区别。它不仅仅关乎“每秒能处理多少请求”（QPS），更关乎“能同时维持多少活跃连接”、“每条连接上的消息吞吐效率如何”以及“连接本身的生命周期是否健康”。

因此，一次专业的WebSocket接口压力测试，就像给后端服务做一次全面的“深度体检”。它需要系统性地评估三个核心指标：连接保活能力、消息吞吐量和并发连接数。这三个指标相互关联，又各有侧重。连接保活考验的是服务在长时间空闲或网络波动下的稳定性；消息吞吐量衡量的是服务处理业务数据流的效率；而并发连接数则直接定义了服务的容量天花板。只有把这三点都测明白，我们才能心中有数，知道服务在真实生产环境中的表现边界在哪里，从而进行有针对性的优化和扩容。

2. 测试环境与工具选型：从零搭建可复现的压测战场

工欲善其事，必先利其器。WebSocket压测工具的选择，直接决定了测试结果的准确性和可操作性。市面上工具众多，我们需要根据测试场景的复杂度、团队技术栈和成本进行权衡。

2.1 主流压测工具横向对比

对于WebSocket压测，我们通常有几类选择：专业的压测平台、开源命令行工具、以及基于代码自研的压测脚本。

1. Apache JMeter这是最广为人知的开源压测工具，通过安装WebSocket Samplers插件即可支持WebSocket。它的优势在于图形化界面，可以方便地组织测试计划、配置参数、查看聚合报告。对于测试用例固定、需要非技术人员参与查看结果的场景比较友好。但其资源消耗较大，单机难以模拟超高并发（如数万连接），且对于复杂交互逻辑（如根据服务器响应决定下一步发送什么消息）的配置略显繁琐。

2. Gatling一个基于Scala的高性能负载测试框架。它采用DSL（领域特定语言）编写测试脚本，代码即配置，版本管理方便。Gatling对WebSocket有原生支持，性能极高，单机可以轻松模拟上万并发。它的测试报告非常专业和详细。缺点是学习曲线稍陡，需要熟悉其DSL语法，更适合开发人员使用。

3. k6一个新兴的、开发者友好的开源负载测试工具，使用JavaScript编写测试脚本。它同样原生支持WebSocket，并且设计理念现代化，易于集成到CI/CD流程中。脚本编写直观，性能也不错。对于前端或Node.js技术栈的团队来说，上手非常快。

4. 自研脚本（使用Python的websockets库或Node.js的ws库）这是最灵活的方式。你可以完全控制连接的生命周期、消息的发送逻辑、以及异常处理。例如，你可以轻松模拟“连接建立后，先订阅A主题，收到推送后再根据内容决定是否发布B消息”这种复杂业务场景。这种方式适合对压测有深度定制化需求的团队，但需要自行实现数据统计、报告生成等功能，前期投入较大。

我的选择与理由：对于需要快速验证、且测试逻辑相对标准的场景（如纯连接保活、固定频率发送消息），我会推荐使用k6。它的脚本简洁，性能足够，报告清晰。而对于需要模拟复杂业务交互、或者追求极限性能的场景，我会选择用Python + websockets + asyncio自研脚本。Python的异步生态非常成熟，可以让我们用少量代码就实现高并发的WebSocket客户端，并且能精细地控制每一个连接的行为，便于插入各种自定义的断言和监控点。

2.2 测试环境搭建要点

压测环境必须独立于生产环境和开发环境，通常我们需要搭建一套专有的压测环境。

服务器端（SUT, System Under Test）：

环境隔离：使用Docker或Kubernetes部署一个与生产环境配置（CPU、内存、JVM参数等）完全一致的待测服务实例。务必确保网络策略允许压测机访问。
监控埋点：在启动服务时，必须开启并暴露关键监控指标。对于JVM应用，这包括GC频率与时长、堆内存使用情况、线程池状态（活跃线程数、队列大小）。此外，还需要应用层监控，如当前活跃WebSocket连接数、消息处理队列长度、平均消息处理耗时等。Prometheus + Grafana是这套监控体系的黄金组合。
日志级别：将日志级别调整到WARN或ERROR，避免压测时产生大量INFO日志拖慢磁盘I/O，影响测试结果。但需要确保连接关闭、异常错误等关键日志能被记录。

压测客户端（Load Generator）：

资源充足：压测机本身的性能必须远高于被测服务，不能成为瓶颈。需要关注CPU、内存、网络带宽以及文件描述符数量限制（ulimit -n）。模拟大量并发连接时，需要大幅提高压测机的最大文件描述符限制（例如设置为65535或更高）。
网络优化：确保压测机与被测服务器处于同一局域网或低延迟的网络环境中，排除网络抖动对结果的影响。如果压测机也需要模拟公网客户端，可以考虑使用不同地域的云服务器。
工具安装：根据选型安装对应工具。例如，选择k6则直接下载二进制包；选择自研Python脚本，则需要创建虚拟环境并安装websockets,aiohttp,asyncio等库。

注意：一个常见的“坑”是忽略了压测机本身的端口限制。当一台机器需要建立数万个到同一目标（服务器IP:Port）的TCP连接时，会大量占用本地端口（每个连接一个本地端口）。默认的本地临时端口范围可能只有两三万个，这会导致“Cannot assign requested address”错误。你需要通过sysctl命令调整net.ipv4.ip_local_port_range参数，扩大端口范围。

3. 核心测试一：连接保活能力测试

连接保活测试，顾名思义，就是检验WebSocket服务在长时间维持大量空闲或低活跃度连接时的稳定性。这听起来简单，却是问题的高发区。

3.1 测试目标与场景设计

核心目标：

验证内存管理：长时间保持数万空闲连接，观察服务进程的内存占用是否线性增长或发生泄漏。
检验心跳机制：测试服务端与客户端预设的心跳（Ping/Pong）机制是否正常工作，能否及时检测并清理“僵尸连接”（死连接）。
评估超时配置：验证服务的各类超时参数（如读超时、写超时、空闲超时）是否合理，是否会误杀健康连接。

测试场景设计：我们需要设计不同“压力”的连接保活场景：

场景A（纯空闲连接）：批量建立N个WebSocket连接，建立成功后，客户端不发送任何应用数据，仅依靠TCP层或WebSocket层的心跳保活。持续运行数小时甚至数十小时。
场景B（低频心跳连接）：建立连接后，客户端以较低频率（如每30秒或60秒）向服务器发送一次特定的Ping消息或业务心跳包，服务器回复Pong或心跳响应。
场景C（网络抖动模拟）：在维持连接的过程中，随机对一部分客户端网络进行短时中断（如使用tc命令模拟网络丢包、延迟），观察服务端能否正确处理重连或清理断连。

3.2 实施步骤与关键监控

以使用Python自研脚本测试“纯空闲连接”场景为例：

import asyncio import websockets import time from contextlib import AsyncExitStack async def single_websocket_client(uri, client_id, stats_dict): """单个WebSocket客户端，仅连接，不发送数据""" try: async with websockets.connect(uri, ping_interval=20, ping_timeout=10) as ws: stats_dict['connected'] += 1 print(f"Client {client_id} connected. Total: {stats_dict['connected']}") # 保持连接，不做任何事，直到被外部取消 await asyncio.Future() # 创建一个永远等待的Future except Exception as e: stats_dict['errors'] += 1 print(f"Client {client_id} error: {e}") async def main(): uri = "ws://your-test-server:port/ws-path" concurrent_connections = 10000 stats = {'connected': 0, 'errors': 0} tasks = [] print(f"Starting to create {concurrent_connections} idle connections...") start_time = time.time() async with AsyncExitStack() as stack: for i in range(concurrent_connections): task = asyncio.create_task(single_websocket_client(uri, i, stats)) tasks.append(task) # 控制连接建立速率，避免瞬间洪峰冲垮服务 if i % 100 == 0: await asyncio.sleep(0.1) # 保持所有连接一段时间（例如2小时） await asyncio.sleep(7200) # 取消所有任务，断开连接 for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) duration = time.time() - start_time print(f"Test finished. Duration: {duration:.2f}s") print(f"Successfully connected: {stats['connected']}") print(f"Connection errors: {stats['errors']}") if __name__ == "__main__": asyncio.run(main())

关键监控项（服务端）：

内存使用量（RSS）：通过top或ps命令，或Prometheus的process_resident_memory_bytes指标，观察其增长曲线。健康的曲线应该是在连接建立初期快速上升，随后趋于平稳，呈一条水平线。如果内存持续缓慢增长，则存在内存泄漏嫌疑。
活跃连接数：与你建立的客户端连接数对比，确保数量一致且稳定。
GC活动：频繁的Full GC会导致服务暂停（Stop-The-World），影响其他连接。监控GC次数和耗时，确保在稳定期GC活动平缓。
文件描述符数量：确保服务没有耗尽文件描述符。

实操心得：

连接建立速率控制：脚本中的await asyncio.sleep(0.1)非常关键。一次性发起数万个TCP连接请求，会对服务器造成SYN洪水攻击般的压力，可能导致部分连接失败，这并非服务容量的真实体现。缓慢递增（Ramp-up）的连接建立方式更能模拟真实场景，也更容易让问题暴露。
“僵尸连接”检测：除了依赖心跳，可以在测试后期，随机挑选几个客户端，尝试发送一条业务消息，看是否能收到服务器响应。如果收不到，说明这个连接在服务端可能已经“僵死”，但未被清理。
关注TIME_WAIT：测试结束后，客户端大量断开连接，会在客户端机器上产生大量TIME_WAIT状态的TCP连接。这是正常的TCP四次挥手过程，但堆积过多可能暂时影响客户端发起新连接。可以通过调整net.ipv4.tcp_tw_reuse等参数来优化。

4. 核心测试二：消息吞吐量测试

消息吞吐量测试关注的是连接建立后，服务处理业务数据流的能力。它回答的问题是：在给定的连接数下，服务每秒能收发多少条消息？平均延迟是多少？

4.1 测试模型与指标定义

吞吐量测试需要定义清晰的测试模型：

消息大小：模拟真实业务消息，例如1KB的JSON文本、10KB的图片元数据包等。需要测试不同消息大小下的吞吐表现。
发送模式：
- 广播：一个客户端发送消息，服务器将其转发给所有其他连接（或特定分组）。测试服务器的多路分发能力。
- 点对点：客户端A定向发送消息给客户端B。测试服务器的路由查找和定向推送能力。
- 请求-响应：客户端发送一条消息，必须等待服务器返回特定响应后才发送下一条。测试服务的同步处理能力。
发送频率：固定频率（如每秒1条）或极限频率（客户端尽可能快地发送，即“背压测试”）。

核心指标：

TPS（Transactions Per Second）：每秒成功完成的事务数。一个事务可以是“发送一条消息并收到确认”。
平均延迟（Average Latency）：从客户端消息发出到收到响应的平均时间。
延迟百分位数（P95, P99 Latency）：例如P99延迟为50ms，意味着99%的请求都在50ms内完成。这个指标比平均延迟更能反映尾部用户体验，对于实时系统至关重要。
错误率：消息发送失败或超时的比例。

4.2 使用k6进行吞吐量测试实战

以下是一个使用k6测试“请求-响应”模式吞吐量的脚本示例。我们假设服务端接口是，客户端发送一个{"type":"echo", "data":"..."}格式的消息，服务端会原样返回。

import { WebSocket } from 'k6/ws'; import { check, sleep } from 'k6'; import { Rate } from 'k6/metrics'; // 自定义指标：错误率 const errorRate = new Rate('errors'); export const options = { stages: [ { duration: '30s', target: 100 }, // 在30秒内逐步增加到100个虚拟用户（连接） { duration: '2m', target: 100 }, // 在100个连接下持续压测2分钟 { duration: '30s', target: 0 }, // 在30秒内逐步降级到0 ], }; export default function () { const url = 'ws://your-test-server:port/ws-path'; const params = { tags: { my_tag: 'websocket_test' } }; // 建立连接 const response = ws.connect(url, params, function (socket) { socket.on('open', function open() { console.log(`VU ${__VU}: connected`); // 设置消息处理函数 socket.on('message', function (data) { try { const msg = JSON.parse(data); // 验证收到的消息是否是刚才发送的回声 const checkResult = check(msg, { 'received echo correctly': (m) => m.data === `test_payload_${__VU}`, }); if (!checkResult) { errorRate.add(1); console.error(`VU ${__VU}: Echo mismatch!`); } } catch (e) { errorRate.add(1); console.error(`VU ${__VU}: Failed to parse message`, e); } }); // 每隔100毫秒发送一条消息，持续发送20次 let count = 0; const intervalId = setInterval(() => { if (count >= 20) { clearInterval(intervalId); socket.close(); return; } const payload = JSON.stringify({ type: 'echo', data: `test_payload_${__VU}_${count}`, timestamp: Date.now(), }); socket.send(payload); count++; }, 100); }); socket.on('error', function (e) { errorRate.add(1); console.error(`VU ${__VU}: WebSocket error`, e.error()); }); socket.on('close', function () { console.log(`VU ${__VU}: disconnected`); }); }); // 检查连接是否成功建立 check(response, { 'status is 101': (r) => r && r.status === 101 }); }

运行测试：k6 run websocket_throughput.js

结果分析：k6会生成一份详细的HTML报告，我们需要重点关注：

http_reqs和iteration_duration：可以换算成TPS和平均延迟。
自定义的errors率：确保错误率在可接受范围内（例如<0.1%）。
在grafana中查看服务端的监控：在压测期间，CPU使用率、消息处理线程池的活跃线程数、消息队列长度是否健康。如果队列持续增长，说明消费者处理速度跟不上生产速度，是潜在的瓶颈。

注意事项：

避免“发后即忘”：上面的例子是等待响应的模式。如果你测试的是单向广播（客户端只发不收），那么TPS可能会非常高，但此时更要关注服务端的消息堆积情况。
消息序列化开销：在真实业务中，消息的JSON序列化/反序列化可能成为CPU热点。压测时使用的消息结构应尽量贴近生产环境。
背压（Backpressure）测试：可以尝试让客户端以极限速度发送消息（去掉setInterval，用循环狂发），观察服务端在过载情况下的表现：是连接断开、消息丢弃，还是响应延迟急剧上升？这有助于定义服务的过载保护策略。

5. 核心测试三：并发连接数极限测试

这是最“暴力”的一项测试，目标是找到服务所能支撑的最大并发连接数。这个数字不仅受限于应用代码，更受限于操作系统配置、服务器硬件和中间件限制。

5.1 测试策略与瓶颈预判

测试策略通常是阶梯式递增：以一定的步长（如每次增加1000连接）逐步增加并发连接数，在每个阶梯上稳定运行一段时间（如5分钟），观察系统指标。直到出现以下任一情况，则认为达到极限：

新建连接失败率超过阈值（如5%）。
已有连接开始大量异常断开。
服务端内存耗尽，或CPU持续100%无法恢复。
平均响应延迟超过可接受范围（如从50ms飙升到2s）。

在测试开始前，我们就需要预判并调整可能的瓶颈点：

服务端瓶颈点：

操作系统级别：
- 文件描述符限制：ulimit -n。需要同时调整服务进程和压测客户端的限制。cat /proc/<pid>/limits可以查看进程实际限制。
- TCP端口范围：net.ipv4.ip_local_port_range（影响客户端）。
- TCP连接内存：net.ipv4.tcp_mem,net.ipv4.tcp_rmem,net.ipv4.tcp_wmem。连接数极高时，可能需要调大这些缓冲区。
- 最大打开文件数：fs.file-max。
中间件/框架级别：
- Web服务器连接数：如Nginx的worker_connections。
- 应用服务器线程池/连接池：如Tomcat的maxConnections，Netty的EventLoop线程数。
应用级别：
- 内存：每个连接都会占用一定的内存（Socket缓冲区、Session对象等）。Java应用尤其要关注堆内存设置（-Xmx）和直接内存（如果用了Netty）。
- 线程资源：如果用的是同步IO模型（如每个连接一个线程），线程数将是主要限制。异步模型（如Netty, Node.js）在这方面有巨大优势。

5.2 实施与瓶颈分析实战

我们使用一个更简单的Python脚本来进行连接数冲刺测试，它只关心“能否连上”和“能否保持”。

import asyncio import websockets import time import sys async def connect_and_hold(uri, client_id, success_counter, error_counter, stop_event): """尝试连接并保持，直到收到停止信号""" try: # 设置较短的超时，快速失败 async with websockets.connect(uri, ping_interval=None, close_timeout=5) as ws: success_counter[0] += 1 if success_counter[0] % 1000 == 0: print(f"[{time.strftime('%H:%M:%S')}] Successfully connected: {success_counter[0]}") # 等待停止事件 await stop_event.wait() # 收到停止信号后，正常关闭 await ws.close() except Exception as e: error_counter[0] += 1 # 可以记录错误类型，如连接拒绝、超时等 if error_counter[0] % 100 == 0: print(f"[{time.strftime('%H:%M:%S')}] Connection errors: {error_counter[0]}, Last error: {e}") async def ramp_up_test(uri, target_connections, ramp_up_step=500, step_duration=30): """阶梯式增加连接测试""" all_tasks = [] success_counter = [0] error_counter = [0] stop_event = asyncio.Event() print(f"Starting ramp-up test to {target_connections} connections...") start_connections = 0 try: while start_connections < target_connections: step_target = min(start_connections + ramp_up_step, target_connections) print(f"\n--- Ramping up from {start_connections} to {step_target} connections ---") # 创建本批次任务 for i in range(start_connections, step_target): task = asyncio.create_task(connect_and_hold(uri, i, success_counter, error_counter, stop_event)) all_tasks.append(task) # 控制建立速率 if len(all_tasks) % 100 == 0: await asyncio.sleep(0.05) # 每100个连接暂停一下 start_connections = step_target # 等待本批次稳定一段时间 print(f"Waiting for {step_duration} seconds to stabilize...") await asyncio.sleep(step_duration) # 打印当前状态 print(f"Current Status: Success={success_counter[0]}, Errors={error_counter[0]}, Active Tasks={len([t for t in all_tasks if not t.done()])}") # 如果错误率已经很高，提前终止测试 total_attempts = success_counter[0] + error_counter[0] if total_attempts > 0 and error_counter[0] / total_attempts > 0.05: print("Error rate exceeded 5%, stopping test.") break # 达到目标后，保持一段时间观察 print(f"\n--- Reached target ({success_counter[0]} successful connections). Holding for 300s ---") await asyncio.sleep(300) except KeyboardInterrupt: print("\nTest interrupted by user.") finally: # 触发停止事件，让所有客户端优雅关闭 print("\nInitiating graceful shutdown...") stop_event.set() # 等待所有任务结束 await asyncio.gather(*all_tasks, return_exceptions=True) print(f"\nFinal Result: Successful connections: {success_counter[0]}, Total errors: {error_counter[0]}") if __name__ == "__main__": server_uri = "ws://your-test-server:port/ws-path" asyncio.run(ramp_up_test(server_uri, target_connections=20000, ramp_up_step=1000, step_duration=60))

测试过程中的监控与瓶颈定位：

连接数卡在某个数字上不去：
- 现象：成功连接数在达到例如10000后停滞，错误增多，多为“Connection refused”或“Timeout”。
- 排查：
  - 服务端：立刻检查服务进程的日志，看是否有“Too many open files”错误。执行ss -s或netstat -an | grep ESTABLISHED | wc -l确认连接数。检查应用框架的连接数配置。
  - 客户端：检查ss -s的TCP: timewait计数是否异常高。可能是本地端口耗尽。
- 解决：根据排查结果，调整对应的系统参数或应用配置。
连接数上去后，内存持续增长直至OOM（Out Of Memory）：
- 现象：连接数随时间推移成功达到目标，但服务端内存占用曲线持续上扬，最终进程被系统杀死。
- 排查：这极有可能是内存泄漏。每个WebSocket连接在服务端都应该对应一个Session或Channel对象。如果连接断开后，这些对象没有被垃圾回收器正确释放，就会产生泄漏。
- 解决：使用jmap（Java）或heapdump等工具，在内存增长期间和增长后分别获取堆内存快照，使用MAT或JProfiler等工具对比分析，找出持有这些失效连接对象的“GC Root”。常见原因包括：未正确移除全局Map中的引用、监听器未取消注册、使用了强引用的缓存等。
连接数高时，CPU占用率100%：
- 现象：连接建立后，即使没有消息收发，服务进程CPU也居高不下。
- 排查：使用top -Hp <pid>查看进程内哪个线程CPU高，再用jstack（Java）或pstack获取线程栈，分析热点代码。常见原因：空轮询（epoll bug）、心跳逻辑有bug导致循环执行、日志框架在低级别下疯狂输出。
- 解决：修复对应的代码逻辑。

6. 综合场景测试与结果分析

单一维度的测试能发现问题，但真实的生产负载往往是混合的。因此，在完成上述三个核心测试后，我们需要设计一个综合场景测试，模拟真实用户行为。

6.1 设计混合负载模型

假设我们为一个在线协作白板应用进行压测，其用户行为可能包括：

用户上线/下线（连接建立/断开）。
用户大部分时间在观看（长连接空闲，偶尔接收他人绘制消息）。
少数用户频繁进行绘制操作（高频发送小消息）。
偶尔有用户上传图片（发送大消息）。

我们可以用以下比例来建模一个虚拟用户的行为（在一个持续5分钟的会话内）：

连接建立后，有80%的概率进入“低频接收”模式：每10秒发送一次心跳，随机接收消息。
有15%的概率进入“高频绘制”模式：每秒发送1-5条绘制指令消息。
有5%的概率在会话中触发一次“上传图片”事件：发送一个50KB的消息。
所有连接在5分钟后统一断开。

使用Python的asyncio可以很好地模拟这种异构行为。我们需要为每个虚拟客户端维护一个独立的状态机，并根据概率随机决定其行为。

6.2 结果整合与报告输出

所有测试完成后，我们需要将结果整合成一份清晰的报告。报告不应只是数据的罗列，而应有分析和结论。

报告核心结构：

测试概述：目标、环境、工具、时间。
容量上限结论：
- 在可接受的延迟（P99 < 100ms）和错误率（<0.1%）下，系统支持的最大并发连接数为X。
- 在Y个并发连接下，系统的消息吞吐量上限为Z TPS（区分消息大小）。
- 系统能够稳定保持X个空闲连接至少N小时，内存增长曲线平稳。
资源消耗模型：
- 给出一个经验公式，例如：总内存预估 ≈ 基础内存 + 每个连接 * 50KB + 每秒消息数 * 处理开销。
- 给出CPU消耗与连接数、TPS的关系曲线图。
关键瓶颈与优化建议：
- 瓶颈1：在连接数达到12000时，受限于worker_connections配置。建议：将Nginx的worker_connections从10240调整为20480。
- 瓶颈2：在广播消息TPS超过5000时，发现单线程分发成为瓶颈。建议：考虑将广播任务拆分为多个子任务，放入线程池并行处理，或使用Redis Pub/Sub进行水平扩展。
- 风险点：在72小时保活测试中，发现内存每小时缓慢增长约2MB，疑似存在微小内存泄漏。建议：使用分析工具进一步定位。
监控仪表板截图：附上测试关键阶段的Grafana仪表板截图，包括连接数、内存、CPU、TPS、延迟百分位数的曲线图，让结论有据可查。
附录：详细的测试脚本、配置参数变更记录、原始数据链接。