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

基于Netty的高并发WebSocket连接管理与性能优化实践指南

news 2025/8/18 10:15:43

cover

基于Netty的高并发WebSocket连接管理与性能优化实践指南

高并发WebSocket 连接在实际生产环境中应用广泛，如在线聊天、推送通知、实时监控等场景。本文将以实战经验分享的形式，结合真实业务场景，介绍如何基于 Netty 完整地规划、实现、调优高并发 WebSocket 服务。

业务场景描述
技术选型过程
实现方案详解
踩过的坑与解决方案
总结与最佳实践

1. 业务场景描述

在某在线教育平台中，需要为成千上万名学生提供实时互动功能，教师端可以推送题目、答案解析，学生端实时接收并提交答题结果，平台要求消息延迟控制在50ms 以内。并发连接数峰值可达50K-100K。

核心需求：

高并发连接管理：需保持长连接，及时推送消息。
低延迟传输：确保消息在毫秒级内送达。
平滑扩缩容：保证集群节点健康时，负载均衡可扩展。
可靠性与稳健性：连接断开后可重连，消息不丢失。

2. 技术选型过程

针对以上需求，评估主流方案：

Spring Boot + WebSocket（Tomcat/NIO）：易用，但在超高并发下 GC 和线程调度开销较大。
Undertow：轻量高效，但社区文档与生态支持有限。
Netty：高性能网络框架，事件驱动、零拷贝、支持自定义协议，社区成熟。

最终选型基于 Netty 实现高并发 WebSocket，借助其高效的 I/O 模型和灵活的 Pipeline 设计，同时结合 Redis 集群做在线用户状态管理、RabbitMQ 做持久化通道消息投递。

3. 实现方案详解

3.1 项目结构

websocket-netty/
├── pom.xml
├── src/main/java/com/example/ws/
│   ├── WebSocketServer.java
│   ├── handler/
│   │   ├── WebSocketServerInitializer.java
│   │   ├── WebSocketFrameHandler.java
│   ├── codec/
│   │   └── JsonFrameCodec.java
│   ├── manager/
│   │   └── ChannelManager.java
│   └── config/
│       └── NettyConfig.java
└── src/main/resources/application.yml

3.2 核心配置示例（application.yml）

netty:port: 8080boss-threads: 2         # boss 线程池大小worker-threads: 16      # worker 线程池大小
redis:host: redis-clusterport: 6379
rabbitmq:host: rabbitmq-serverport: 5672username: userpassword: pass

3.3 Netty 服务端启动

public class WebSocketServer {public static void main(String[] args) throws InterruptedException {NettyConfig config = NettyConfig.load();EventLoopGroup bossGroup = new NioEventLoopGroup(config.getBossThreads());EventLoopGroup workerGroup = new NioEventLoopGroup(config.getWorkerThreads());try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new WebSocketServerInitializer());ChannelFuture f = b.bind(config.getPort()).sync();System.out.println("WebSocket 服务已启动，端口：" + config.getPort());f.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

3.4 Pipeline 初始化

public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) {ChannelPipeline p = ch.pipeline();p.addLast(new HttpServerCodec());p.addLast(new HttpObjectAggregator(65536));p.addLast(new WebSocketServerProtocolHandler("/ws", null, true));p.addLast(new JsonFrameCodec());          // 自定义 JSON 编解码器p.addLast(new WebSocketFrameHandler());  // 业务处理}
}

3.5 在线连接管理

使用单例 ChannelManager 管理所有活动连接：

public class ChannelManager {// 存储 userId -> Channel 映射private final ConcurrentMap<String, Channel> channels = new ConcurrentHashMap<>();private static final ChannelManager INSTANCE = new ChannelManager();private ChannelManager() {}public static ChannelManager getInstance() { return INSTANCE; }public void add(String userId, Channel ch) {channels.put(userId, ch);}public void remove(String userId) {channels.remove(userId);}public Channel get(String userId) {return channels.get(userId);}public Collection<Channel> all() {return channels.values();}
}

3.6 消息推送示例

public class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {// 假设客户端首条报文为 {"type":"CONNECT","userId":"u123"}JsonNode json = JsonFrameCodec.parse(msg.text());String type = json.get("type").asText();if ("CONNECT".equals(type)) {String userId = json.get("userId").asText();ChannelManager.getInstance().add(userId, ctx.channel());return;}// 普通消息推送String to = json.get("to").asText();Channel target = ChannelManager.getInstance().get(to);if (target != null && target.isActive()) {target.writeAndFlush(new TextWebSocketFrame(msg.text()));}}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {// 移除断开连接的 ChannelChannelManager.getInstance().remove(ctx.channel().id().asShortText());}
}

3.7 性能测试与优化

使用 wrk2 进行并发压力测试：

wrk2 -t8 -c10000 -d60s --latency http://localhost:8080/ws

调整 Netty 线程池大小：根据 CPU 核数 + 1 进行实验调整。
启用 -XX:+UseG1GC 并设置 -XX:MaxGCPauseMillis=100 降低 GC 停顿。
合理设置 writeBufferHighWaterMark/LowWaterMark，防止大流量写缓存积累。

测试结果： | 并发连接数 | 平均延迟(ms) | 最大延迟(ms) | | --------- | ---------- | ---------- | | 10K | 12 | 45 | | 50K | 28 | 90 |

4. 踩过的坑与解决方案

Channel 泄漏：部分异常分支未调用 ctx.close()，导致连接未释放。
解决：在 exceptionCaught 中统一关闭并移除 Channel。
ByteBuf 内存泄漏：忘记在自定义 JsonFrameCodec 中释放 ByteBuf。
解决：在 decode/encode 方法中手动调用 ReferenceCountUtil.release(in)。
线程阻塞：在 Handler 中执行耗时操作（如 DB 写入）。
解决：使用异步线程池或 EventExecutorGroup 分离业务逻辑。