1. 项目概述与核心价值最近在做一个需要高并发、低延迟实时数据推送的项目比如在线协作白板、实时行情或者大规模聊天室传统的HTTP轮询或者长轮询方案在性能和资源消耗上很快就遇到了瓶颈。这时候WebSocket就成了不二之选。但直接用Spring Boot内置的Tomcat容器提供的WebSocket支持在面对几千上万的并发连接时内存和线程消耗会非常可观性能曲线也不够理想。于是我把目光投向了Netty——这个以高性能和高并发处理能力著称的NIO框架。用SpringBoot整合Netty来搭建WebSocket服务本质上就是把SpringBoot的便捷开发与Netty的网络性能优势结合起来打造一个既能快速开发又能扛住压力的实时通信后端。这个组合能解决的核心问题很明确为需要海量双向实时通信的应用提供一个稳定、高效、可扩展的后端服务。它适合那些对延迟敏感、连接数多、且消息交互频繁的场景。无论是刚接触Netty想了解如何与SpringBoot集成的开发者还是正在为现有实时服务性能发愁的架构师都可以从这个方案里找到参考价值。接下来我会从设计思路、代码实现、到最后的压测验证完整地走一遍这个流程并附上可以直接运行的Python测试脚本。2. 技术选型与整体架构设计2.1 为什么是SpringBoot Netty选择这个技术栈背后有非常实际的工程考量。首先SpringBoot极大地简化了项目的初始配置和依赖管理让我们能快速搭建起一个具备完整生态如依赖注入、外部化配置、健康检查的应用骨架。如果纯手写Netty服务光是处理配置、日志、监控这些周边设施就要花不少功夫。而Netty的核心优势在于其基于事件驱动和异步非阻塞NIO的架构。对于WebSocket这种长连接服务每个连接在大部分时间是空闲的等待消息。基于BIO的Tomcat容器会为每个连接分配一个线程在连接数暴涨时线程上下文切换的开销和内存占用会成为主要瓶颈。Netty则通过少量的线程如BossGroup和WorkerGroup来处理海量连接的事件连接建立、数据读取、数据写出连接空闲时不会占用计算资源这使得它在高并发场景下的资源利用率和吞吐量有数量级的提升。简单来说SpringBoot负责“过日子”应用生命周期、配置、业务Bean管理Netty负责“赚钱养家”高效处理网络I/O。两者通过Spring的生命周期事件如ApplicationRunner或CommandLineRunner进行桥接让Netty服务在Spring应用启动后自动运行并在关闭时优雅退出。2.2 服务端核心架构拆解一个基于Netty的WebSocket服务器其核心架构可以分解为以下几个层次引导层Bootstrap这是Netty服务的启动入口。我们需要创建两个EventLoopGroup通常称为bossGroup和workerGroup。bossGroup负责接收客户端的连接请求workerGroup负责处理已建立连接的数据读写。然后组装一个ServerBootstrap将这两个Group、Channel类型NIO以及最重要的ChannelPipeline配置关联起来。管道处理器链ChannelPipeline这是Netty处理逻辑的核心。数据在Netty中被封装为ByteBuf像流水一样经过Pipeline中的一个个处理器ChannelHandler。对于WebSocket我们需要一组特定的处理器来解析HTTP协议升级请求并管理WebSocket帧。HttpServerCodec负责HTTP请求的编解码。HttpObjectAggregator将HTTP请求的多个部分如分块传输聚合成一个完整的FullHttpRequest对象这对于处理WebSocket握手请求一个完整的HTTP请求是必须的。WebSocketServerProtocolHandler这是关键。它自动处理WebSocket握手Handshake的繁琐细节包括验证Upgrade头、计算Sec-WebSocket-Accept响应等。握手成功后它会将HTTP协议升级为WebSocket协议后续的数据都将以WebSocket帧的形式传递。自定义业务处理器通常继承SimpleChannelInboundHandlerTextWebSocketFrame或BinaryWebSocketFrame。在这里我们处理连接建立、断开、收到文本或二进制消息、以及异常事件并编写具体的业务逻辑如消息广播、点对点发送等。连接与会话管理这是业务层面的核心。我们需要一个全局的结构来管理所有在线的WebSocket连接Netty的Channel。通常使用一个线程安全的ConcurrentHashMap以用户ID或设备ID为Key对应的Channel为Value进行存储。当需要向特定用户发送消息时就可以从这个Map中找到对应的Channel进行操作。2.3 项目依赖配置Maven在pom.xml中我们需要引入以下核心依赖。版本号建议使用较新的稳定版。dependencies !-- Spring Boot Starter -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !-- Spring Boot 配置处理器可选便于配置提示 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-configuration-processor/artifactId optionaltrue/optional /dependency !-- Netty 所有依赖 -- dependency groupIdio.netty/groupId artifactIdnetty-all/artifactId version4.1.100.Final/version /dependency !-- 工具类如JSON处理 -- dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- 测试 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies注意Netty的netty-all依赖已经包含了我们需要的所有Netty模块如编解码器、WebSocket支持等直接引入它是最简单的方式。确保版本一致避免因版本冲突导致WebSocketServerProtocolHandler等类找不到。3. 服务端核心代码实现详解3.1 配置类与参数外部化首先我们将Netty服务器的配置如端口、线程数提取到application.yml中便于不同环境切换。# application.yml netty: websocket: port: 8080 # WebSocket服务端口 boss-thread-count: 1 # BossGroup线程数通常1个足够 worker-thread-count: 8 # WorkerGroup线程数建议设置为CPU核心数*2 path: /ws # WebSocket握手请求的路径然后创建一个配置类来绑定这些属性import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; Component ConfigurationProperties(prefix netty.websocket) Data public class NettyWebSocketConfig { private int port; private int bossThreadCount; private int workerThreadCount; private String path; }3.2 WebSocket服务启动器NettyServer这是整合SpringBoot与Netty的关键类。我们实现ApplicationRunner接口让Netty服务在SpringBoot启动完成后自动运行。import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.stream.ChunkedWriteHandler; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; Component Slf4j RequiredArgsConstructor public class NettyWebSocketServer implements ApplicationRunner { private final NettyWebSocketConfig nettyConfig; private final WebSocketMessageHandler webSocketMessageHandler; // 自定义业务处理器 private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; Override public void run(ApplicationArguments args) throws Exception { log.info(开始启动Netty WebSocket服务器端口{}路径{}, nettyConfig.getPort(), nettyConfig.getPath()); // 1. 创建线程组 bossGroup new NioEventLoopGroup(nettyConfig.getBossThreadCount()); workerGroup new NioEventLoopGroup(nettyConfig.getWorkerThreadCount()); try { // 2. 创建服务器启动引导 ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // 使用NIO传输通道 .handler(new LoggingHandler(LogLevel.INFO)) // 为BossGroup添加日志处理器 .childHandler(new ChannelInitializerSocketChannel() { // 为每个新连接设置Pipeline Override protected void initChannel(SocketChannel ch) { ch.pipeline() // 添加HTTP编解码器 .addLast(new HttpServerCodec()) // 支持大数据流写入如文件 .addLast(new ChunkedWriteHandler()) // 将HTTP消息的多个部分聚合为一个完整的FullHttpRequest/FullHttpResponse .addLast(new HttpObjectAggregator(65536)) // 最大聚合内容长度64KB // WebSocket协议处理器负责握手、心跳(ping/pong)、关闭帧处理 .addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true)) // 自定义的业务处理器处理文本消息等 .addLast(webSocketMessageHandler); } }) // TCP参数配置 .option(ChannelOption.SO_BACKLOG, 128) // 连接队列大小 .childOption(ChannelOption.SO_KEEPALIVE, true); // 开启TCP心跳 // 3. 绑定端口同步等待成功 ChannelFuture future bootstrap.bind(nettyConfig.getPort()).sync(); log.info(Netty WebSocket服务器启动成功监听端口{}, nettyConfig.getPort()); // 4. 等待服务端监听端口关闭阻塞直到Channel关闭 future.channel().closeFuture().sync(); } finally { // 优雅关闭释放线程池资源 shutdownGracefully(); } } PreDestroy public void shutdownGracefully() { if (workerGroup ! null) { workerGroup.shutdownGracefully(); } if (bossGroup ! null) { bossGroup.shutdownGracefully(); } log.info(Netty WebSocket服务器线程组已关闭); } }关键点解析HttpObjectAggregator(65536)这个参数很重要它定义了单个HTTP请求内容的最大长度。WebSocket握手是一个HTTP请求如果客户端发送的握手请求头或附加数据过大超过这个限制连接会被直接关闭。需要根据实际业务调整但一般64KB足够。WebSocketServerProtocolHandler第三个参数allowExtensions设为true允许WebSocket扩展。它帮我们处理了所有标准WebSocket帧如Ping/Pong心跳、关闭帧我们只需要关心文本或二进制数据帧。SO_BACKLOG当服务器处理连接的速度跟不上请求到达的速度时请求会被放入这个队列。在高并发场景下适当调大此值如1024可以应对瞬间的连接洪峰但也不宜过大。3.3 自定义业务消息处理器WebSocketMessageHandler这是业务逻辑的核心。我们需要管理连接并处理各种事件。import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.util.concurrent.ConcurrentHashMap; Component Slf4j public class WebSocketMessageHandler extends SimpleChannelInboundHandlerWebSocketFrame { // 存储用户ID与Channel的映射关系。实际项目中用户ID可能从握手请求的token中解析。 private static final ConcurrentHashMapString, Channel userChannelMap new ConcurrentHashMap(); // 存储Channel与用户ID的映射用于连接断开时清理 private static final ConcurrentHashMapChannel, String channelUserMap new ConcurrentHashMap(); Override protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) { // 判断是否是文本帧我们这里只处理文本消息 if (frame instanceof TextWebSocketFrame) { String requestText ((TextWebSocketFrame) frame).text(); log.info(收到来自客户端[{}]的消息{}, ctx.channel().id(), requestText); // TODO: 解析消息内容例如JSON格式{userId:123,type:chat,content:hello} // 这里简单做回声测试 String response 服务器回声: requestText; ctx.channel().writeAndFlush(new TextWebSocketFrame(response)); } else { // 不支持二进制帧或其他帧可以关闭连接或忽略 String message 不支持的数据帧类型: frame.getClass().getName(); ctx.channel().writeAndFlush(new TextWebSocketFrame(message)); log.warn(message); } } /** * 客户端连接成功时触发 */ Override public void handlerAdded(ChannelHandlerContext ctx) { Channel channel ctx.channel(); String channelId channel.id().asShortText(); log.info(客户端连接加入Channel ID: {}, channelId); // 这里通常没有用户信息可以等第一个认证消息过来后再绑定 } /** * 客户端连接断开时触发 */ Override public void handlerRemoved(ChannelHandlerContext ctx) { Channel channel ctx.channel(); String userId channelUserMap.remove(channel); if (userId ! null) { userChannelMap.remove(userId); log.info(用户[{}]断开连接Channel ID: {}, userId, channel.id().asShortText()); } else { log.info(未认证连接断开Channel ID: {}, channel.id().asShortText()); } } /** * 处理用户事件用于实现心跳检测 */ Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.READER_IDLE) { // 读空闲即一段时间内没有收到客户端消息可以认为连接已失效关闭它 log.warn(Channel[{}]读空闲超时关闭连接, ctx.channel().id().asShortText()); ctx.channel().close(); } // 还可以处理 WRITER_IDLE写空闲和 ALL_IDLE全部空闲 } else { super.userEventTriggered(ctx, evt); } } /** * 发生异常时触发 */ Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { log.error(WebSocket连接发生异常Channel ID: {}, ctx.channel().id().asShortText(), cause); ctx.close(); } /** * 绑定用户ID与Channel业务方法可在处理认证消息后调用 */ public static void bindUserToChannel(String userId, Channel channel) { Channel oldChannel userChannelMap.put(userId, channel); if (oldChannel ! null oldChannel.isActive()) { // 如果该用户已有其他活跃连接可以强制关闭旧的实现单设备登录 oldChannel.close(); log.info(用户[{}]的旧连接已被强制关闭, userId); } channelUserMap.put(channel, userId); log.info(用户[{}]绑定到Channel[{}], userId, channel.id().asShortText()); } /** * 根据用户ID发送消息业务方法 */ public static void sendMessageToUser(String userId, String message) { Channel channel userChannelMap.get(userId); if (channel ! null channel.isActive()) { channel.writeAndFlush(new TextWebSocketFrame(message)); log.debug(向用户[{}]发送消息: {}, userId, message); } else { log.warn(用户[{}]不在线或连接已失效消息发送失败: {}, userId, message); // TODO: 可以在这里实现离线消息存储 } } /** * 广播消息给所有在线用户 */ public static void broadcastMessage(String message) { userChannelMap.forEach((userId, channel) - { if (channel.isActive()) { channel.writeAndFlush(new TextWebSocketFrame(message)); } }); log.debug(广播消息: {}, message); } }业务逻辑关键点与避坑指南连接管理使用两个ConcurrentHashMap进行双向映射管理是常见做法。务必注意在handlerRemoved和exceptionCaught中清理映射防止内存泄漏。Channel的isActive()方法可以判断连接是否依然有效。用户认证WebSocket协议本身不处理认证。常见的做法是在握手阶段通过URL参数如ws://host:port/ws?tokenxxx传递Token然后在自定义的ChannelHandler可以加在WebSocketServerProtocolHandler之前中解析Token并验证。验证通过后再调用bindUserToChannel方法绑定用户。如果验证失败可以直接关闭连接ctx.close()。心跳与空闲检测为了及时清理死连接如客户端异常崩溃、网络断开必须加入心跳机制。Netty提供了IdleStateHandler。我们需要在Pipeline中添加它并在userEventTriggered方法中处理超时事件。例如设置读空闲时间为60秒如果60秒内没收到任何消息就断开连接。// 在ChannelInitializer的initChannel方法中添加 .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)) // 读超时60秒消息格式实际业务中消息通常是结构化的JSON。建议定义一个统一的消息格式包含消息类型type、发送者from、接收者to、内容content、时间戳timestamp等字段。在channelRead0中解析JSON根据type字段路由到不同的业务处理方法。线程安全ChannelHandler的方法会被Netty的I/O线程调用。我们的业务逻辑如访问数据库、调用其他服务如果比较耗时绝对不能阻塞I/O线程否则会严重影响网络处理性能。必须将耗时操作提交到业务线程池EventExecutorGroup中执行。可以使用ctx.channel().eventLoop().execute(Runnable task)提交到当前Channel关联的I/O线程或者使用自定义的业务线程池。3.4 集成心跳检测在NettyWebSocketServer的ChannelInitializer中在添加业务处理器之前插入IdleStateHandler。// 在 initChannel 方法内添加在业务处理器之前 ch.pipeline() .addLast(new HttpServerCodec()) .addLast(new ChunkedWriteHandler()) .addLast(new HttpObjectAggregator(65536)) .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)) // 读空闲60秒触发 .addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true)) .addLast(webSocketMessageHandler);这样当连接60秒内没有读取到任何数据时会触发IdleStateEvent.READER_IDLE事件我们在WebSocketMessageHandler的userEventTriggered方法中已经处理了关闭逻辑。4. 性能测试方案设计与Python脚本实现服务搭建好了性能到底如何我们需要一个能模拟大量并发客户端进行连接、发送和接收消息的测试工具。这里选择Python的websockets库因为它异步性能好编写脚本方便。4.1 测试环境与目标测试机一台配置较好的Linux服务器如4核8G。服务部署将SpringBootNetty服务打包成Jar在该服务器上运行。测试目标连接容量服务器能稳定维持多少个WebSocket连接消息吞吐量在特定连接数下每秒能处理多少条消息TPS消息延迟从客户端发送消息到收到服务器回声平均延迟是多少资源消耗在不同压力下服务器的CPU、内存占用情况。4.2 Python异步测试脚本详解我们需要编写两个脚本一个用于模拟大量静默连接测试连接容量另一个用于模拟活跃连接发送消息测试吞吐和延迟。脚本一stress_connect.py- 纯连接压力测试这个脚本只建立连接不发送消息用于测试服务器维持海量空闲连接的能力。import asyncio import websockets import time import sys async def connect_and_hold(uri, client_id): 连接服务器并保持连接 try: async with websockets.connect(uri) as websocket: print(f客户端 {client_id} 连接成功) # 保持连接直到被外部取消或出错 await asyncio.Future() # 创建一个永远等待的Future except Exception as e: print(f客户端 {client_id} 连接失败: {e}) async def main(): if len(sys.argv) ! 4: print(用法: python stress_connect.py 服务器地址 连接数 并发批次大小) print(示例: python stress_connect.py ws://localhost:8080/ws 10000 500) sys.exit(1) server_address sys.argv[1] total_clients int(sys.argv[2]) batch_size int(sys.argv[3]) # 每批创建的连接数避免瞬间创建过多 tasks [] connected_count 0 start_time time.time() for i in range(total_clients): task asyncio.create_task(connect_and_hold(f{server_address}?clientId{i}, i)) tasks.append(task) connected_count 1 # 控制并发创建速度每批完成后等待一小会儿 if len(tasks) batch_size: # 等待当前批次的所有连接尝试完成成功或失败 done, pending await asyncio.wait(tasks, return_whenasyncio.FIRST_COMPLETED) # 这里简单处理实际可以更精细地统计成功数 print(f已发起 {connected_count} 个连接...) # 清空已完成的任务保留未完成的正在连接的 tasks list(pending) await asyncio.sleep(0.1) # 短暂间隔给系统喘息时间 # 等待所有剩余任务 if tasks: await asyncio.wait(tasks) end_time time.time() print(f\n连接测试完成。) print(f目标连接数: {total_clients}) print(f总耗时: {end_time - start_time:.2f} 秒) # 注意这里统计的是发起的连接数实际成功数需要更复杂的逻辑统计 if __name__ __main__: asyncio.run(main())使用方式python stress_connect.py ws://你的服务器IP:8080/ws 5000 200。这个命令会尝试建立5000个连接每批200个。脚本二stress_message.py- 消息吞吐与延迟测试这个脚本模拟客户端连接后持续向服务器发送消息并等待回声用于测试消息处理能力。import asyncio import websockets import time import sys import json from collections import deque # 全局统计变量 send_count 0 receive_count 0 latencies deque(maxlen10000) # 保存最近10000次延迟用于计算 async def message_client(uri, client_id, message_interval): 单个客户端连接、定时发送消息、统计延迟 global send_count, receive_count try: async with websockets.connect(uri) as websocket: print(f客户端 {client_id} 连接成功开始发送消息) while True: # 构造消息 message json.dumps({ clientId: client_id, timestamp: time.time(), data: Hello, Server! }) send_time time.time() # 发送消息 await websocket.send(message) send_count 1 # 等待回声 try: # 设置接收超时防止服务器无响应导致客户端挂起 echo await asyncio.wait_for(websocket.recv(), timeout5.0) recv_time time.time() latency (recv_time - send_time) * 1000 # 转换为毫秒 latencies.append(latency) receive_count 1 # print(f客户端 {client_id} 收到回声延迟: {latency:.2f}ms) except asyncio.TimeoutError: print(f客户端 {client_id} 接收超时) break # 按指定间隔发送下一条消息 await asyncio.sleep(message_interval) except Exception as e: print(f客户端 {client_id} 出错: {e}) async def print_stats(): 定时打印统计信息 while True: await asyncio.sleep(5) # 每5秒打印一次 if latencies: avg_latency sum(latencies) / len(latencies) min_latency min(latencies) max_latency max(latencies) else: avg_latency min_latency max_latency 0 print(f\n--- 统计信息 ---) print(f发送消息总数: {send_count}) print(f接收消息总数: {receive_count}) print(f消息丢失率: {(send_count - receive_count)/max(send_count,1)*100:.2f}%) print(f平均延迟: {avg_latency:.2f}ms, 最小延迟: {min_latency:.2f}ms, 最大延迟: {max_latency:.2f}ms) print(f当前采样延迟数: {len(latencies)}) print(f---\n) async def main(): if len(sys.argv) ! 5: print(用法: python stress_message.py 服务器地址 客户端数 消息间隔(秒) 测试时长(秒)) print(示例: python stress_message.py ws://localhost:8080/ws 100 0.1 60) sys.exit(1) server_address sys.argv[1] client_count int(sys.argv[2]) message_interval float(sys.argv[3]) # 每个客户端发送消息的间隔 test_duration int(sys.argv[4]) # 启动统计打印任务 stats_task asyncio.create_task(print_stats()) # 创建所有客户端任务 tasks [] for i in range(client_count): task asyncio.create_task(message_client(f{server_address}?clientId{i}, i, message_interval)) tasks.append(task) await asyncio.sleep(0.01) # 稍微错开连接启动时间 # 运行指定时长 print(f测试开始将持续 {test_duration} 秒...) await asyncio.sleep(test_duration) # 测试结束取消所有客户端任务 for task in tasks: task.cancel() # 等待所有任务被取消 await asyncio.gather(*tasks, return_exceptionsTrue) stats_task.cancel() # 最终统计 print(\n *50) print(测试结束最终统计) if latencies: avg_latency sum(latencies) / len(latencies) print(f总发送/接收: {send_count}/{receive_count}) print(f平均延迟: {avg_latency:.2f}ms) print(*50) if __name__ __main__: asyncio.run(main())使用方式python stress_message.py ws://你的服务器IP:8080/ws 200 0.05 120。这个命令会启动200个客户端每个客户端每0.05秒即每秒20条发送一条消息持续测试120秒。4.3 测试执行与监控要点环境准备确保测试机和服务端不在同一台机器避免资源竞争影响结果。在服务端机器上使用java -jar -Xms512m -Xmx512m your-app.jar启动服务并监控其进程资源用top或htop。系统限制调整在Linux服务器上需要调整系统参数以支持大量连接。文件描述符限制ulimit -n 65535或 修改/etc/security/limits.conf。TCP参数调整修改/etc/sysctl.conf增加net.core.somaxconn连接队列、net.ipv4.tcp_tw_reuse和net.ipv4.tcp_tw_recycle快速回收TIME_WAIT连接注意高版本内核已移除tcp_tw_recycle建议用tcp_tw_reuse然后执行sysctl -p生效。分阶段测试基准测试先用少量连接如100和低频率消息验证功能正常。容量测试运行stress_connect.py逐步增加连接数如1000, 5000, 10000观察服务器内存占用和CPU使用率是否平稳。使用netstat -an | grep :8080 | wc -l查看实际连接数。压力测试运行stress_message.py在稳定的最大连接数下逐步提高消息发送频率观察TPS和延迟的变化曲线找到性能拐点。结果分析关注几个关键指标内存增长连接建立后内存是否稳定有无内存泄漏持续增长CPU使用率在消息吞吐测试中CPU是瓶颈吗是用户态业务逻辑高还是系统态网络I/O高延迟分布平均延迟是否在可接受范围如50ms延迟的方差抖动大不大错误率连接失败率、消息丢失率是否接近05. 性能优化与生产环境考量通过测试我们可能会发现瓶颈。以下是一些常见的优化方向和生产环境必须考虑的事项。5.1 服务端性能优化点JVM参数调优-Xms和-Xmx设置堆内存初始值和最大值建议设为相同值避免运行时调整带来的性能波动。根据连接数估算每个Netty Channel及其关联对象在空闲时占用内存不大几十KB但海量连接下总内存可观。例如10万连接可能需要2-4G的堆内存。-XX:UseG1GC对于需要低延迟的应用G1垃圾收集器通常比Parallel GC有更好的表现尤其是大内存场景下。-XX:MaxGCPauseMillis设置G1的目标最大GC停顿时间。Netty参数调优SO_BACKLOG根据并发连接建立的峰值调整。ChannelOption.ALLOCATOR使用PooledByteBufAllocator.DEFAULTNetty 4.1默认就是池化分配器可以显著减少ByteBuf分配和回收的开销。EventLoopGroup线程数workerGroup线程数并非越多越好。一般设置为CPU核心数的1-2倍。过多的I/O线程会增加上下文切换开销。可以通过监控I/O线程的繁忙程度来调整。业务逻辑优化避免阻塞I/O线程重申一遍所有数据库查询、远程服务调用等阻塞操作必须提交到业务线程池。批处理与合并对于高频的、可以合并的消息如实时位置更新可以考虑在客户端或服务端做少量缓冲和合并减少网络包数量和服务器处理次数。连接预热在服务启动后可以先建立一部分“预热”连接促使JVM完成JIT编译和相关类的加载使服务在承受真实流量前达到最佳性能。5.2 生产环境部署与高可用多实例与负载均衡单机性能总有上限。需要通过负载均衡器如Nginx、HAProxy、云厂商的LB将WebSocket连接分发到多个后端Netty服务实例。这里有个关键点WebSocket是长连接需要负载均衡器支持会话保持粘性会话否则同一个用户的后续请求可能被路由到不同的后端导致状态丢失。Nginx的ip_hash或hash $connection_id指令可以实现。健康检查负载均衡器需要对后端服务进行健康检查。除了TCP端口检查最好提供一个HTTP端点如/health由Netty服务端额外绑定一个简单的HTTP服务可以使用Netty本身也可以在一个单独的端口启动一个Spring Boot Actuator来响应。优雅下线在服务重启或缩容时需要先通知负载均衡器摘掉该节点然后等待一段时间如30秒让现有连接处理完业务后再关闭Netty服务器。这可以通过监听Spring的ContextClosedEvent在事件中先拒绝新连接再逐步关闭现有连接来实现。监控与告警必须建立完善的监控体系。应用层监控在线连接数、消息收发速率、不同消息类型的处理延迟P99 P95。系统层监控服务器的CPU、内存、网络带宽、TCP连接状态ESTABLISHED,TIME_WAIT。JVM层监控GC频率和耗时、堆内存使用情况。可以使用Prometheus Grafana进行采集和可视化并在关键指标异常时触发告警。5.3 常见问题排查实录在实际部署和测试中你可能会遇到以下问题连接数达到一定数量后无法再建立新连接可能原因服务器进程的文件描述符File Descriptor耗尽。排查使用ulimit -n查看当前限制使用lsof -p pid | wc -l查看进程已使用的FD数。解决调整系统级和进程级的文件描述符限制。大量TIME_WAIT状态的连接现象netstat -an | grep TIME_WAIT数量非常多可能导致端口耗尽。原因主动关闭连接的一方服务器或客户端会进入TIME_WAIT状态等待2MSL通常60秒。解决调整内核参数net.ipv4.tcp_tw_reuse和net.ipv4.tcp_tw_recycle谨慎使用或者确保由客户端主动断开连接服务器尽量不主动close。内存缓慢增长最终OOM可能原因内存泄漏。最常见的是Channel或相关对象没有被正确释放未从userChannelMap中移除。排查使用jmap -histo:live pid观察对象数量或使用MAT等工具分析堆转储文件查看Channel、ChannelHandlerContext等Netty对象的累积情况。解决仔细检查handlerRemoved和exceptionCaught中的资源清理逻辑确保所有路径都能执行到。CPU使用率异常高可能原因业务逻辑中有死循环或低效算法。频繁的GC垃圾回收。Netty的I/O线程在处理耗时业务被阻塞。排查使用top -Hp pid查看哪个线程CPU高再用jstack pid获取线程栈分析热点代码。解决优化业务代码将阻塞操作移出I/O线程。测试脚本报错“too many open files”原因测试客户端机器本身的文件描述符不够用。解决同样需要调整测试机的ulimit -n值。搭建一个高性能的WebSocket服务编码只是第一步。从设计、实现、测试到最终上线运维每一个环节都需要仔细考量。SpringBootNetty的组合给了我们一个强大的起点但要让它真正在生产环境中稳定、高效地运行离不开持续的调优、完善的监控和清晰的故障排查思路。希望这篇从零到一再到性能压测的完整分享能帮你少踩一些坑。 SpringBoot整合Netty构建高性能WebSocket服务：架构设计与压测实战
1. 项目概述与核心价值最近在做一个需要高并发、低延迟实时数据推送的项目比如在线协作白板、实时行情或者大规模聊天室传统的HTTP轮询或者长轮询方案在性能和资源消耗上很快就遇到了瓶颈。这时候WebSocket就成了不二之选。但直接用Spring Boot内置的Tomcat容器提供的WebSocket支持在面对几千上万的并发连接时内存和线程消耗会非常可观性能曲线也不够理想。于是我把目光投向了Netty——这个以高性能和高并发处理能力著称的NIO框架。用SpringBoot整合Netty来搭建WebSocket服务本质上就是把SpringBoot的便捷开发与Netty的网络性能优势结合起来打造一个既能快速开发又能扛住压力的实时通信后端。这个组合能解决的核心问题很明确为需要海量双向实时通信的应用提供一个稳定、高效、可扩展的后端服务。它适合那些对延迟敏感、连接数多、且消息交互频繁的场景。无论是刚接触Netty想了解如何与SpringBoot集成的开发者还是正在为现有实时服务性能发愁的架构师都可以从这个方案里找到参考价值。接下来我会从设计思路、代码实现、到最后的压测验证完整地走一遍这个流程并附上可以直接运行的Python测试脚本。2. 技术选型与整体架构设计2.1 为什么是SpringBoot Netty选择这个技术栈背后有非常实际的工程考量。首先SpringBoot极大地简化了项目的初始配置和依赖管理让我们能快速搭建起一个具备完整生态如依赖注入、外部化配置、健康检查的应用骨架。如果纯手写Netty服务光是处理配置、日志、监控这些周边设施就要花不少功夫。而Netty的核心优势在于其基于事件驱动和异步非阻塞NIO的架构。对于WebSocket这种长连接服务每个连接在大部分时间是空闲的等待消息。基于BIO的Tomcat容器会为每个连接分配一个线程在连接数暴涨时线程上下文切换的开销和内存占用会成为主要瓶颈。Netty则通过少量的线程如BossGroup和WorkerGroup来处理海量连接的事件连接建立、数据读取、数据写出连接空闲时不会占用计算资源这使得它在高并发场景下的资源利用率和吞吐量有数量级的提升。简单来说SpringBoot负责“过日子”应用生命周期、配置、业务Bean管理Netty负责“赚钱养家”高效处理网络I/O。两者通过Spring的生命周期事件如ApplicationRunner或CommandLineRunner进行桥接让Netty服务在Spring应用启动后自动运行并在关闭时优雅退出。2.2 服务端核心架构拆解一个基于Netty的WebSocket服务器其核心架构可以分解为以下几个层次引导层Bootstrap这是Netty服务的启动入口。我们需要创建两个EventLoopGroup通常称为bossGroup和workerGroup。bossGroup负责接收客户端的连接请求workerGroup负责处理已建立连接的数据读写。然后组装一个ServerBootstrap将这两个Group、Channel类型NIO以及最重要的ChannelPipeline配置关联起来。管道处理器链ChannelPipeline这是Netty处理逻辑的核心。数据在Netty中被封装为ByteBuf像流水一样经过Pipeline中的一个个处理器ChannelHandler。对于WebSocket我们需要一组特定的处理器来解析HTTP协议升级请求并管理WebSocket帧。HttpServerCodec负责HTTP请求的编解码。HttpObjectAggregator将HTTP请求的多个部分如分块传输聚合成一个完整的FullHttpRequest对象这对于处理WebSocket握手请求一个完整的HTTP请求是必须的。WebSocketServerProtocolHandler这是关键。它自动处理WebSocket握手Handshake的繁琐细节包括验证Upgrade头、计算Sec-WebSocket-Accept响应等。握手成功后它会将HTTP协议升级为WebSocket协议后续的数据都将以WebSocket帧的形式传递。自定义业务处理器通常继承SimpleChannelInboundHandlerTextWebSocketFrame或BinaryWebSocketFrame。在这里我们处理连接建立、断开、收到文本或二进制消息、以及异常事件并编写具体的业务逻辑如消息广播、点对点发送等。连接与会话管理这是业务层面的核心。我们需要一个全局的结构来管理所有在线的WebSocket连接Netty的Channel。通常使用一个线程安全的ConcurrentHashMap以用户ID或设备ID为Key对应的Channel为Value进行存储。当需要向特定用户发送消息时就可以从这个Map中找到对应的Channel进行操作。2.3 项目依赖配置Maven在pom.xml中我们需要引入以下核心依赖。版本号建议使用较新的稳定版。dependencies !-- Spring Boot Starter -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !-- Spring Boot 配置处理器可选便于配置提示 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-configuration-processor/artifactId optionaltrue/optional /dependency !-- Netty 所有依赖 -- dependency groupIdio.netty/groupId artifactIdnetty-all/artifactId version4.1.100.Final/version /dependency !-- 工具类如JSON处理 -- dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- 测试 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies注意Netty的netty-all依赖已经包含了我们需要的所有Netty模块如编解码器、WebSocket支持等直接引入它是最简单的方式。确保版本一致避免因版本冲突导致WebSocketServerProtocolHandler等类找不到。3. 服务端核心代码实现详解3.1 配置类与参数外部化首先我们将Netty服务器的配置如端口、线程数提取到application.yml中便于不同环境切换。# application.yml netty: websocket: port: 8080 # WebSocket服务端口 boss-thread-count: 1 # BossGroup线程数通常1个足够 worker-thread-count: 8 # WorkerGroup线程数建议设置为CPU核心数*2 path: /ws # WebSocket握手请求的路径然后创建一个配置类来绑定这些属性import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; Component ConfigurationProperties(prefix netty.websocket) Data public class NettyWebSocketConfig { private int port; private int bossThreadCount; private int workerThreadCount; private String path; }3.2 WebSocket服务启动器NettyServer这是整合SpringBoot与Netty的关键类。我们实现ApplicationRunner接口让Netty服务在SpringBoot启动完成后自动运行。import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.stream.ChunkedWriteHandler; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; Component Slf4j RequiredArgsConstructor public class NettyWebSocketServer implements ApplicationRunner { private final NettyWebSocketConfig nettyConfig; private final WebSocketMessageHandler webSocketMessageHandler; // 自定义业务处理器 private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; Override public void run(ApplicationArguments args) throws Exception { log.info(开始启动Netty WebSocket服务器端口{}路径{}, nettyConfig.getPort(), nettyConfig.getPath()); // 1. 创建线程组 bossGroup new NioEventLoopGroup(nettyConfig.getBossThreadCount()); workerGroup new NioEventLoopGroup(nettyConfig.getWorkerThreadCount()); try { // 2. 创建服务器启动引导 ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // 使用NIO传输通道 .handler(new LoggingHandler(LogLevel.INFO)) // 为BossGroup添加日志处理器 .childHandler(new ChannelInitializerSocketChannel() { // 为每个新连接设置Pipeline Override protected void initChannel(SocketChannel ch) { ch.pipeline() // 添加HTTP编解码器 .addLast(new HttpServerCodec()) // 支持大数据流写入如文件 .addLast(new ChunkedWriteHandler()) // 将HTTP消息的多个部分聚合为一个完整的FullHttpRequest/FullHttpResponse .addLast(new HttpObjectAggregator(65536)) // 最大聚合内容长度64KB // WebSocket协议处理器负责握手、心跳(ping/pong)、关闭帧处理 .addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true)) // 自定义的业务处理器处理文本消息等 .addLast(webSocketMessageHandler); } }) // TCP参数配置 .option(ChannelOption.SO_BACKLOG, 128) // 连接队列大小 .childOption(ChannelOption.SO_KEEPALIVE, true); // 开启TCP心跳 // 3. 绑定端口同步等待成功 ChannelFuture future bootstrap.bind(nettyConfig.getPort()).sync(); log.info(Netty WebSocket服务器启动成功监听端口{}, nettyConfig.getPort()); // 4. 等待服务端监听端口关闭阻塞直到Channel关闭 future.channel().closeFuture().sync(); } finally { // 优雅关闭释放线程池资源 shutdownGracefully(); } } PreDestroy public void shutdownGracefully() { if (workerGroup ! null) { workerGroup.shutdownGracefully(); } if (bossGroup ! null) { bossGroup.shutdownGracefully(); } log.info(Netty WebSocket服务器线程组已关闭); } }关键点解析HttpObjectAggregator(65536)这个参数很重要它定义了单个HTTP请求内容的最大长度。WebSocket握手是一个HTTP请求如果客户端发送的握手请求头或附加数据过大超过这个限制连接会被直接关闭。需要根据实际业务调整但一般64KB足够。WebSocketServerProtocolHandler第三个参数allowExtensions设为true允许WebSocket扩展。它帮我们处理了所有标准WebSocket帧如Ping/Pong心跳、关闭帧我们只需要关心文本或二进制数据帧。SO_BACKLOG当服务器处理连接的速度跟不上请求到达的速度时请求会被放入这个队列。在高并发场景下适当调大此值如1024可以应对瞬间的连接洪峰但也不宜过大。3.3 自定义业务消息处理器WebSocketMessageHandler这是业务逻辑的核心。我们需要管理连接并处理各种事件。import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.util.concurrent.ConcurrentHashMap; Component Slf4j public class WebSocketMessageHandler extends SimpleChannelInboundHandlerWebSocketFrame { // 存储用户ID与Channel的映射关系。实际项目中用户ID可能从握手请求的token中解析。 private static final ConcurrentHashMapString, Channel userChannelMap new ConcurrentHashMap(); // 存储Channel与用户ID的映射用于连接断开时清理 private static final ConcurrentHashMapChannel, String channelUserMap new ConcurrentHashMap(); Override protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) { // 判断是否是文本帧我们这里只处理文本消息 if (frame instanceof TextWebSocketFrame) { String requestText ((TextWebSocketFrame) frame).text(); log.info(收到来自客户端[{}]的消息{}, ctx.channel().id(), requestText); // TODO: 解析消息内容例如JSON格式{userId:123,type:chat,content:hello} // 这里简单做回声测试 String response 服务器回声: requestText; ctx.channel().writeAndFlush(new TextWebSocketFrame(response)); } else { // 不支持二进制帧或其他帧可以关闭连接或忽略 String message 不支持的数据帧类型: frame.getClass().getName(); ctx.channel().writeAndFlush(new TextWebSocketFrame(message)); log.warn(message); } } /** * 客户端连接成功时触发 */ Override public void handlerAdded(ChannelHandlerContext ctx) { Channel channel ctx.channel(); String channelId channel.id().asShortText(); log.info(客户端连接加入Channel ID: {}, channelId); // 这里通常没有用户信息可以等第一个认证消息过来后再绑定 } /** * 客户端连接断开时触发 */ Override public void handlerRemoved(ChannelHandlerContext ctx) { Channel channel ctx.channel(); String userId channelUserMap.remove(channel); if (userId ! null) { userChannelMap.remove(userId); log.info(用户[{}]断开连接Channel ID: {}, userId, channel.id().asShortText()); } else { log.info(未认证连接断开Channel ID: {}, channel.id().asShortText()); } } /** * 处理用户事件用于实现心跳检测 */ Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.READER_IDLE) { // 读空闲即一段时间内没有收到客户端消息可以认为连接已失效关闭它 log.warn(Channel[{}]读空闲超时关闭连接, ctx.channel().id().asShortText()); ctx.channel().close(); } // 还可以处理 WRITER_IDLE写空闲和 ALL_IDLE全部空闲 } else { super.userEventTriggered(ctx, evt); } } /** * 发生异常时触发 */ Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { log.error(WebSocket连接发生异常Channel ID: {}, ctx.channel().id().asShortText(), cause); ctx.close(); } /** * 绑定用户ID与Channel业务方法可在处理认证消息后调用 */ public static void bindUserToChannel(String userId, Channel channel) { Channel oldChannel userChannelMap.put(userId, channel); if (oldChannel ! null oldChannel.isActive()) { // 如果该用户已有其他活跃连接可以强制关闭旧的实现单设备登录 oldChannel.close(); log.info(用户[{}]的旧连接已被强制关闭, userId); } channelUserMap.put(channel, userId); log.info(用户[{}]绑定到Channel[{}], userId, channel.id().asShortText()); } /** * 根据用户ID发送消息业务方法 */ public static void sendMessageToUser(String userId, String message) { Channel channel userChannelMap.get(userId); if (channel ! null channel.isActive()) { channel.writeAndFlush(new TextWebSocketFrame(message)); log.debug(向用户[{}]发送消息: {}, userId, message); } else { log.warn(用户[{}]不在线或连接已失效消息发送失败: {}, userId, message); // TODO: 可以在这里实现离线消息存储 } } /** * 广播消息给所有在线用户 */ public static void broadcastMessage(String message) { userChannelMap.forEach((userId, channel) - { if (channel.isActive()) { channel.writeAndFlush(new TextWebSocketFrame(message)); } }); log.debug(广播消息: {}, message); } }业务逻辑关键点与避坑指南连接管理使用两个ConcurrentHashMap进行双向映射管理是常见做法。务必注意在handlerRemoved和exceptionCaught中清理映射防止内存泄漏。Channel的isActive()方法可以判断连接是否依然有效。用户认证WebSocket协议本身不处理认证。常见的做法是在握手阶段通过URL参数如ws://host:port/ws?tokenxxx传递Token然后在自定义的ChannelHandler可以加在WebSocketServerProtocolHandler之前中解析Token并验证。验证通过后再调用bindUserToChannel方法绑定用户。如果验证失败可以直接关闭连接ctx.close()。心跳与空闲检测为了及时清理死连接如客户端异常崩溃、网络断开必须加入心跳机制。Netty提供了IdleStateHandler。我们需要在Pipeline中添加它并在userEventTriggered方法中处理超时事件。例如设置读空闲时间为60秒如果60秒内没收到任何消息就断开连接。// 在ChannelInitializer的initChannel方法中添加 .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)) // 读超时60秒消息格式实际业务中消息通常是结构化的JSON。建议定义一个统一的消息格式包含消息类型type、发送者from、接收者to、内容content、时间戳timestamp等字段。在channelRead0中解析JSON根据type字段路由到不同的业务处理方法。线程安全ChannelHandler的方法会被Netty的I/O线程调用。我们的业务逻辑如访问数据库、调用其他服务如果比较耗时绝对不能阻塞I/O线程否则会严重影响网络处理性能。必须将耗时操作提交到业务线程池EventExecutorGroup中执行。可以使用ctx.channel().eventLoop().execute(Runnable task)提交到当前Channel关联的I/O线程或者使用自定义的业务线程池。3.4 集成心跳检测在NettyWebSocketServer的ChannelInitializer中在添加业务处理器之前插入IdleStateHandler。// 在 initChannel 方法内添加在业务处理器之前 ch.pipeline() .addLast(new HttpServerCodec()) .addLast(new ChunkedWriteHandler()) .addLast(new HttpObjectAggregator(65536)) .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)) // 读空闲60秒触发 .addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true)) .addLast(webSocketMessageHandler);这样当连接60秒内没有读取到任何数据时会触发IdleStateEvent.READER_IDLE事件我们在WebSocketMessageHandler的userEventTriggered方法中已经处理了关闭逻辑。4. 性能测试方案设计与Python脚本实现服务搭建好了性能到底如何我们需要一个能模拟大量并发客户端进行连接、发送和接收消息的测试工具。这里选择Python的websockets库因为它异步性能好编写脚本方便。4.1 测试环境与目标测试机一台配置较好的Linux服务器如4核8G。服务部署将SpringBootNetty服务打包成Jar在该服务器上运行。测试目标连接容量服务器能稳定维持多少个WebSocket连接消息吞吐量在特定连接数下每秒能处理多少条消息TPS消息延迟从客户端发送消息到收到服务器回声平均延迟是多少资源消耗在不同压力下服务器的CPU、内存占用情况。4.2 Python异步测试脚本详解我们需要编写两个脚本一个用于模拟大量静默连接测试连接容量另一个用于模拟活跃连接发送消息测试吞吐和延迟。脚本一stress_connect.py- 纯连接压力测试这个脚本只建立连接不发送消息用于测试服务器维持海量空闲连接的能力。import asyncio import websockets import time import sys async def connect_and_hold(uri, client_id): 连接服务器并保持连接 try: async with websockets.connect(uri) as websocket: print(f客户端 {client_id} 连接成功) # 保持连接直到被外部取消或出错 await asyncio.Future() # 创建一个永远等待的Future except Exception as e: print(f客户端 {client_id} 连接失败: {e}) async def main(): if len(sys.argv) ! 4: print(用法: python stress_connect.py 服务器地址 连接数 并发批次大小) print(示例: python stress_connect.py ws://localhost:8080/ws 10000 500) sys.exit(1) server_address sys.argv[1] total_clients int(sys.argv[2]) batch_size int(sys.argv[3]) # 每批创建的连接数避免瞬间创建过多 tasks [] connected_count 0 start_time time.time() for i in range(total_clients): task asyncio.create_task(connect_and_hold(f{server_address}?clientId{i}, i)) tasks.append(task) connected_count 1 # 控制并发创建速度每批完成后等待一小会儿 if len(tasks) batch_size: # 等待当前批次的所有连接尝试完成成功或失败 done, pending await asyncio.wait(tasks, return_whenasyncio.FIRST_COMPLETED) # 这里简单处理实际可以更精细地统计成功数 print(f已发起 {connected_count} 个连接...) # 清空已完成的任务保留未完成的正在连接的 tasks list(pending) await asyncio.sleep(0.1) # 短暂间隔给系统喘息时间 # 等待所有剩余任务 if tasks: await asyncio.wait(tasks) end_time time.time() print(f\n连接测试完成。) print(f目标连接数: {total_clients}) print(f总耗时: {end_time - start_time:.2f} 秒) # 注意这里统计的是发起的连接数实际成功数需要更复杂的逻辑统计 if __name__ __main__: asyncio.run(main())使用方式python stress_connect.py ws://你的服务器IP:8080/ws 5000 200。这个命令会尝试建立5000个连接每批200个。脚本二stress_message.py- 消息吞吐与延迟测试这个脚本模拟客户端连接后持续向服务器发送消息并等待回声用于测试消息处理能力。import asyncio import websockets import time import sys import json from collections import deque # 全局统计变量 send_count 0 receive_count 0 latencies deque(maxlen10000) # 保存最近10000次延迟用于计算 async def message_client(uri, client_id, message_interval): 单个客户端连接、定时发送消息、统计延迟 global send_count, receive_count try: async with websockets.connect(uri) as websocket: print(f客户端 {client_id} 连接成功开始发送消息) while True: # 构造消息 message json.dumps({ clientId: client_id, timestamp: time.time(), data: Hello, Server! }) send_time time.time() # 发送消息 await websocket.send(message) send_count 1 # 等待回声 try: # 设置接收超时防止服务器无响应导致客户端挂起 echo await asyncio.wait_for(websocket.recv(), timeout5.0) recv_time time.time() latency (recv_time - send_time) * 1000 # 转换为毫秒 latencies.append(latency) receive_count 1 # print(f客户端 {client_id} 收到回声延迟: {latency:.2f}ms) except asyncio.TimeoutError: print(f客户端 {client_id} 接收超时) break # 按指定间隔发送下一条消息 await asyncio.sleep(message_interval) except Exception as e: print(f客户端 {client_id} 出错: {e}) async def print_stats(): 定时打印统计信息 while True: await asyncio.sleep(5) # 每5秒打印一次 if latencies: avg_latency sum(latencies) / len(latencies) min_latency min(latencies) max_latency max(latencies) else: avg_latency min_latency max_latency 0 print(f\n--- 统计信息 ---) print(f发送消息总数: {send_count}) print(f接收消息总数: {receive_count}) print(f消息丢失率: {(send_count - receive_count)/max(send_count,1)*100:.2f}%) print(f平均延迟: {avg_latency:.2f}ms, 最小延迟: {min_latency:.2f}ms, 最大延迟: {max_latency:.2f}ms) print(f当前采样延迟数: {len(latencies)}) print(f---\n) async def main(): if len(sys.argv) ! 5: print(用法: python stress_message.py 服务器地址 客户端数 消息间隔(秒) 测试时长(秒)) print(示例: python stress_message.py ws://localhost:8080/ws 100 0.1 60) sys.exit(1) server_address sys.argv[1] client_count int(sys.argv[2]) message_interval float(sys.argv[3]) # 每个客户端发送消息的间隔 test_duration int(sys.argv[4]) # 启动统计打印任务 stats_task asyncio.create_task(print_stats()) # 创建所有客户端任务 tasks [] for i in range(client_count): task asyncio.create_task(message_client(f{server_address}?clientId{i}, i, message_interval)) tasks.append(task) await asyncio.sleep(0.01) # 稍微错开连接启动时间 # 运行指定时长 print(f测试开始将持续 {test_duration} 秒...) await asyncio.sleep(test_duration) # 测试结束取消所有客户端任务 for task in tasks: task.cancel() # 等待所有任务被取消 await asyncio.gather(*tasks, return_exceptionsTrue) stats_task.cancel() # 最终统计 print(\n *50) print(测试结束最终统计) if latencies: avg_latency sum(latencies) / len(latencies) print(f总发送/接收: {send_count}/{receive_count}) print(f平均延迟: {avg_latency:.2f}ms) print(*50) if __name__ __main__: asyncio.run(main())使用方式python stress_message.py ws://你的服务器IP:8080/ws 200 0.05 120。这个命令会启动200个客户端每个客户端每0.05秒即每秒20条发送一条消息持续测试120秒。4.3 测试执行与监控要点环境准备确保测试机和服务端不在同一台机器避免资源竞争影响结果。在服务端机器上使用java -jar -Xms512m -Xmx512m your-app.jar启动服务并监控其进程资源用top或htop。系统限制调整在Linux服务器上需要调整系统参数以支持大量连接。文件描述符限制ulimit -n 65535或 修改/etc/security/limits.conf。TCP参数调整修改/etc/sysctl.conf增加net.core.somaxconn连接队列、net.ipv4.tcp_tw_reuse和net.ipv4.tcp_tw_recycle快速回收TIME_WAIT连接注意高版本内核已移除tcp_tw_recycle建议用tcp_tw_reuse然后执行sysctl -p生效。分阶段测试基准测试先用少量连接如100和低频率消息验证功能正常。容量测试运行stress_connect.py逐步增加连接数如1000, 5000, 10000观察服务器内存占用和CPU使用率是否平稳。使用netstat -an | grep :8080 | wc -l查看实际连接数。压力测试运行stress_message.py在稳定的最大连接数下逐步提高消息发送频率观察TPS和延迟的变化曲线找到性能拐点。结果分析关注几个关键指标内存增长连接建立后内存是否稳定有无内存泄漏持续增长CPU使用率在消息吞吐测试中CPU是瓶颈吗是用户态业务逻辑高还是系统态网络I/O高延迟分布平均延迟是否在可接受范围如50ms延迟的方差抖动大不大错误率连接失败率、消息丢失率是否接近05. 性能优化与生产环境考量通过测试我们可能会发现瓶颈。以下是一些常见的优化方向和生产环境必须考虑的事项。5.1 服务端性能优化点JVM参数调优-Xms和-Xmx设置堆内存初始值和最大值建议设为相同值避免运行时调整带来的性能波动。根据连接数估算每个Netty Channel及其关联对象在空闲时占用内存不大几十KB但海量连接下总内存可观。例如10万连接可能需要2-4G的堆内存。-XX:UseG1GC对于需要低延迟的应用G1垃圾收集器通常比Parallel GC有更好的表现尤其是大内存场景下。-XX:MaxGCPauseMillis设置G1的目标最大GC停顿时间。Netty参数调优SO_BACKLOG根据并发连接建立的峰值调整。ChannelOption.ALLOCATOR使用PooledByteBufAllocator.DEFAULTNetty 4.1默认就是池化分配器可以显著减少ByteBuf分配和回收的开销。EventLoopGroup线程数workerGroup线程数并非越多越好。一般设置为CPU核心数的1-2倍。过多的I/O线程会增加上下文切换开销。可以通过监控I/O线程的繁忙程度来调整。业务逻辑优化避免阻塞I/O线程重申一遍所有数据库查询、远程服务调用等阻塞操作必须提交到业务线程池。批处理与合并对于高频的、可以合并的消息如实时位置更新可以考虑在客户端或服务端做少量缓冲和合并减少网络包数量和服务器处理次数。连接预热在服务启动后可以先建立一部分“预热”连接促使JVM完成JIT编译和相关类的加载使服务在承受真实流量前达到最佳性能。5.2 生产环境部署与高可用多实例与负载均衡单机性能总有上限。需要通过负载均衡器如Nginx、HAProxy、云厂商的LB将WebSocket连接分发到多个后端Netty服务实例。这里有个关键点WebSocket是长连接需要负载均衡器支持会话保持粘性会话否则同一个用户的后续请求可能被路由到不同的后端导致状态丢失。Nginx的ip_hash或hash $connection_id指令可以实现。健康检查负载均衡器需要对后端服务进行健康检查。除了TCP端口检查最好提供一个HTTP端点如/health由Netty服务端额外绑定一个简单的HTTP服务可以使用Netty本身也可以在一个单独的端口启动一个Spring Boot Actuator来响应。优雅下线在服务重启或缩容时需要先通知负载均衡器摘掉该节点然后等待一段时间如30秒让现有连接处理完业务后再关闭Netty服务器。这可以通过监听Spring的ContextClosedEvent在事件中先拒绝新连接再逐步关闭现有连接来实现。监控与告警必须建立完善的监控体系。应用层监控在线连接数、消息收发速率、不同消息类型的处理延迟P99 P95。系统层监控服务器的CPU、内存、网络带宽、TCP连接状态ESTABLISHED,TIME_WAIT。JVM层监控GC频率和耗时、堆内存使用情况。可以使用Prometheus Grafana进行采集和可视化并在关键指标异常时触发告警。5.3 常见问题排查实录在实际部署和测试中你可能会遇到以下问题连接数达到一定数量后无法再建立新连接可能原因服务器进程的文件描述符File Descriptor耗尽。排查使用ulimit -n查看当前限制使用lsof -p pid | wc -l查看进程已使用的FD数。解决调整系统级和进程级的文件描述符限制。大量TIME_WAIT状态的连接现象netstat -an | grep TIME_WAIT数量非常多可能导致端口耗尽。原因主动关闭连接的一方服务器或客户端会进入TIME_WAIT状态等待2MSL通常60秒。解决调整内核参数net.ipv4.tcp_tw_reuse和net.ipv4.tcp_tw_recycle谨慎使用或者确保由客户端主动断开连接服务器尽量不主动close。内存缓慢增长最终OOM可能原因内存泄漏。最常见的是Channel或相关对象没有被正确释放未从userChannelMap中移除。排查使用jmap -histo:live pid观察对象数量或使用MAT等工具分析堆转储文件查看Channel、ChannelHandlerContext等Netty对象的累积情况。解决仔细检查handlerRemoved和exceptionCaught中的资源清理逻辑确保所有路径都能执行到。CPU使用率异常高可能原因业务逻辑中有死循环或低效算法。频繁的GC垃圾回收。Netty的I/O线程在处理耗时业务被阻塞。排查使用top -Hp pid查看哪个线程CPU高再用jstack pid获取线程栈分析热点代码。解决优化业务代码将阻塞操作移出I/O线程。测试脚本报错“too many open files”原因测试客户端机器本身的文件描述符不够用。解决同样需要调整测试机的ulimit -n值。搭建一个高性能的WebSocket服务编码只是第一步。从设计、实现、测试到最终上线运维每一个环节都需要仔细考量。SpringBootNetty的组合给了我们一个强大的起点但要让它真正在生产环境中稳定、高效地运行离不开持续的调优、完善的监控和清晰的故障排查思路。希望这篇从零到一再到性能压测的完整分享能帮你少踩一些坑。