# Netty 连接管理优化：如何支持 10万+ 并发连接？
在 IM 系统中，高并发连接是核心挑战。本文介绍 AQChat 如何通过 Netty 连接管理优化支持 10万+ 并发连接。

## 一、10万+ 并发连接的挑战
### 挑战
1. 内存占用：每个连接约 2-4KB，10万连接约 200-400MB
2. 线程资源：传统 BIO 模型需要 10万线程，资源消耗大
3. 死连接检测：即时清理无效连接，避免资源泄漏
4. 消息广播：房间内消息需要高效广播给所有成员

### 解决方案：使用 Netty NIO 模型 + 连接管理优化

## 二、Boss/Worker 线程模型的选择
Netty 的 Reactor 线程模型

Netty 采用主从 Reactor 多线程模型：
* Boss 线程：负责接收客户端连接
* Worker 线程：处理已建立的 I/O 事件

**代码实现**：
```java
@SpringBootConfiguration
public class AQChatNettyStarter implements InitializingBean {
    
    private void startImApplication() throws InterruptedException {
        // 1. 创建线程组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(aqChatConfig.getBossThreadSize());  // 1个Boss线程
        NioEventLoopGroup workGroup = new NioEventLoopGroup(aqChatConfig.getWorkThreadSize());  // 8个Worker线程
        
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 配置Pipeline
                        ch.pipeline().addLast(new HttpServerCodec());
                        ch.pipeline().addLast(new ChunkedWriteHandler());
                        ch.pipeline().addLast(new HttpObjectAggregator(65535));
                        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/"));
                        ch.pipeline().addLast(new IdleStateHandler(0,0,10, TimeUnit.MINUTES));
                        ch.pipeline().addLast(messageDecoder);
                        ch.pipeline().addLast(messageEncoder);
                        ch.pipeline().addLast(hearBeatHandler);
                        ch.pipeline().addLast(aqChatCommandHandler);
                    }
                });
        
        ChannelFuture channelFuture = bootstrap.bind(aqChatConfig.getWebSocketPort()).sync();
        LOGGER.info("Netty服务启动成功，监听端口：{}", aqChatConfig.getWebSocketPort());
    }
}
```

**配置文件**（`application.yml`）：
```yml
aq-chat:
  websocketPort: 9090
  bossThreadSize: 1   # Boss线程：1个（通常1个足够）
  workThreadSize: 8   # Worker线程：8个（根据CPU核数调整）
```

### 为什么这样配置？
1.Boss 线程：1 个
* 只负责接收连接，不处理 I/O
* 1 个线程可处理大量连接请求
* 过多 Boss 线程无意义

2.Worker 线程：8 个
* 处理 I/O 事件（读、写）
* 建议为 CPU 核数的 1-2 倍
* 8 核 CPU 建议 8-16 个 Worker 线程

**性能对比**：
| **模型** | **线程数** | **10万连接内存** | **适用场景** |
|------|-------|----------|--------|
| BIO | 10万 | 高 | 不适合高并发 |
| NIO（1 Boss + 8 Worker） | 9 | 低 | 适合高并发 |

## 三、ChannelGroup 的使用和优化
### ChannelGroup 的作用
`ChannelGroup` 用于管理一组 `Channel`，支持批量操作（如房间内消息广播）。

**代码实现**：
```java
@Component
public class GlobalChannelHolder {
    
    @Resource
    private MessageBroadcaster messageBroadcaster;
    
    /**
     * 创建房间的ChannelGroup
     */
    public void createChannelGroup(String roomId) {
        // 使用GlobalEventExecutor作为EventExecutor
        ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        messageBroadcaster.addChannelGroup(roomId, channelGroup);
    }
    
    /**
     * 用户加入房间
     */
    public void joinRoom(String roomId, String userId, Channel channel) {
        messageBroadcaster.joinRoom(roomId, userId, (NioSocketChannel) channel);
    }
}
```
```java
@Component
public class MessageBroadcaster {
    
    /**
     * 房间号 -> ChannelGroup 映射
     */
    private final Map<String, ChannelGroup> channelGroupMap = new ConcurrentHashMap<>();
    
    /**
     * 用户 -> 房间 映射
     */
    private final Map<String, String> userRoomMap = new ConcurrentHashMap<>();
    
    /**
     * 加入房间
     */
    public void joinRoom(String roomId, String userId, NioSocketChannel channel) {
        userRoomMap.put(userId, roomId);
        ChannelGroup channelGroup = channelGroupMap.get(roomId);
        if (channelGroup != null) {
            channelGroup.add(channel);  // 添加到ChannelGroup
        }
    }
    
    /**
     * 离开房间
     */
    public String leaveRoom(String userId, NioSocketChannel channel) {
        String roomId = userRoomMap.remove(userId);
        if (roomId != null) {
            ChannelGroup channelGroup = channelGroupMap.get(roomId);
            if (channelGroup != null) {
                channelGroup.remove(channel);  // 从ChannelGroup移除
            }
        }
        return roomId;
    }
    
    /**
     * 广播消息到房间
     */
    public <T extends GeneratedMessageV3> void broadcast(String roomId, T msg) {
        ChannelGroup channelGroup = channelGroupMap.get(roomId);
        if (channelGroup != null) {
            // 批量发送，高效
            channelGroup.writeAndFlush(msg);
        }
    }
}
```

### ChannelGroup 的优势
1. 批量操作：`writeAndFlush()` 可批量发送，比循环发送更高效
2. 自动管理：Channel 关闭时自动从 ChannelGroup 移除
3. 线程安全：内部使用 `ConcurrentHashMap`，支持并发操作

**性能对比**：
| **方式** | **100人房间广播耗时** | **说明** |
|------|----------|------|
| 循环发送 | 10-20ms | 逐个发送，效率低 |
| ChannelGroup | 1-2ms | 批量发送，效率高 |

## 四、心跳机制的设计（10分钟超时）
### 为什么需要心跳？
1. 检测死连接：网络异常时连接可能未正常关闭
2. 保持连接活跃：防止被防火墙/NAT 设备断开
3. 即时释放资源：清理无效连接，释放内存

### 实现方案：`IdleStateHandler` + 自定义心跳处理器
```java
// 1. 在Pipeline中添加IdleStateHandler
ch.pipeline().addLast(new IdleStateHandler(0, 0, 10, TimeUnit.MINUTES));
// 参数说明：
// - readerIdleTime: 读空闲时间（0表示不检测）
// - writerIdleTime: 写空闲时间（0表示不检测）
// - allIdleTime: 读写空闲时间（10分钟）
```
```java
@Component
@ChannelHandler.Sharable
public class HearBeatHandler extends ChannelInboundHandlerAdapter {
    
    @Resource
    private GlobalChannelHolder channelHolder;
    
    @Resource
    private IUserHolder userHolder;
    
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent event) {
            if (event.state() == IdleState.ALL_IDLE) {
                // 10分钟无读写，触发超时
                String userId = (String) ctx.channel().attr(AttributeKey.valueOf(AQBusinessConstant.USER_ID)).get();
                
                if (userId != null) {
                    LOGGER.info("用户{}心跳超时10分钟，发送离线消息并断开连接", userId);
                    
                    // 1. 构建离线消息
                    AQChatMsgProtocol.OfflineMsg offlineMsg = buildOfflineMsg(userId);
                    ctx.writeAndFlush(offlineMsg);
                    
                    // 2. 清理资源
                    userHolder.offline(userId);
                    channelHolder.offline(userId, (NioSocketChannel) ctx.channel());
                }
                
                // 3. 关闭连接
                ctx.channel().close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
```
**客户端心跳响应**：
```java
@Component
public class HeartBeatCmdHandler extends AbstractCmdBaseHandler<AQChatMsgProtocol.HeartBeatCmd> {
    
    @Override
    public void handle(ChannelHandlerContext ctx, AQChatMsgProtocol.HeartBeatCmd cmd) {
        String userId = verifyLogin(ctx);
        if (userId != null) {
            // 更新心跳时间
            ctx.channel().attr(AttributeKey.valueOf(AQBusinessConstant.HEART_BEAT_TIME))
                .set(System.currentTimeMillis());
            
            // 返回心跳响应
            ctx.writeAndFlush(AQChatMsgProtocol.HeartBeatAck.newBuilder()
                .setPong(AQBusinessConstant.HEART_BEAT_ACK)
                .build());
        }
    }
}
```

### 心跳机制的优势
1. 自动检测：`IdleStateHandler` 自动检测空闲连接
2. 及时清理：10 分钟无活动自动断开，释放资源
3. 资源释放：触发 `channelInactive`，清理相关资源

## 五、死连接检测和清理
### 死连接检测的两种方式
1.**心跳超时检测**（主动）：
```java
// HearBeatHandler.java
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
    if (evt instanceof IdleStateEvent event && event.state() == IdleState.ALL_IDLE) {
        // 10分钟无读写，判定为死连接
        cleanupConnection(ctx);
        ctx.channel().close();
    }
}
```
2.**连接断开检测**（被动）
```java
// AQChatCommandHandler.java
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    super.channelInactive(ctx);
    
    String userId = (String) ctx.channel().attr(AttributeKey.valueOf(AQBusinessConstant.USER_ID)).get();
    if (userId != null) {
        LOGGER.info("用户{}断开连接", userId);
        
        // 1. 设置用户离线
        userHolder.offline(userId);
        
        // 2. 移除用户连接
        globalChannelHolder.offline(userId, (NioSocketChannel) ctx.channel());
        
        // 3. 发送离线消息
        mqSendingAgent.sendOfflineMessage(userId);
    }
    
    ctx.close();
}
```
**清理流程**：
```java
public void offline(String userId, NioSocketChannel nioSocketChannel) {
    // 1. 从CHANNELS Map中移除
    remove(userId);
    
    // 2. 从ChannelGroup中移除
    messageBroadcaster.removeChannel(userId, nioSocketChannel);
}
```
```java
public void removeChannel(String userId, NioSocketChannel nioSocketChannel) {
    String roomId = userRoomMap.get(userId);
    if (roomId != null) {
        ChannelGroup channelGroup = channelGroupMap.get(roomId);
        if (channelGroup != null) {
            channelGroup.remove(nioSocketChannel);  // 从ChannelGroup移除
        }
    }
    userRoomMap.remove(userId);  // 从用户房间映射中移除
}
```

## 六、内存优化：及时释放 Channel 引用
### 内存泄漏的原因
1. Channel 引用未释放：`ConcurrentHashMap` 中保留无效 Channel
2. ChannelGroup 未清理：房间解散后 ChaneelGroup 未移除
3. 用户映射未清理：用户离线后映射关系未清除

### 优化措施
1.**及时移除 Channel 引用**
```java
// GlobalChannelHolder.java
private final Map<String, NioSocketChannel> CHANNELS = new ConcurrentHashMap<>();

public NioSocketChannel remove(String userId) {
    // 及时移除，释放内存
    return CHANNELS.remove(userId);
}
```
2.**房间解散时清理 ChannelGroup**
```java
public void isOrNoDissolveTheRoom(RoomInfoDto roomInfoDto) {
    String roomId = roomInfoDto.getRoomId();
    ChannelGroup channelGroup = messageBroadcaster.getChannelGroup(roomId);
    
    // 如果房间为空，清理ChannelGroup
    if (channelGroup == null || channelGroup.isEmpty()) {
        // 移除房间信息
        roomHolder.removeRoomInfo(roomInfoDto.getRoomNo());
        
        // 移除ChannelGroup，释放内存
        messageBroadcaster.removeChannelGroup(roomId);
    }
}
```
3.**用户离线时清理所有映射**
```java
public void offline(String userId, NioSocketChannel nioSocketChannel) {
    // 1. 从CHANNELS移除
    remove(userId);
    
    // 2. 从ChannelGroup移除
    messageBroadcaster.removeChannel(userId, nioSocketChannel);
    
    // 3. 从用户房间映射移除
    messageBroadcaster.remove(userId);
}
```

### 内存优化效果
| **优化项** | **优化前** | **优化后** | **说明** |
|--------|--------|--------|------|
| 死连接内存占用 | 持续增长 | 及时释放 | 10 分钟超时自动清理 |
| ChannelGroup 泄漏 | 房间解散后仍存在 | 及时清理 | 房间为空时自动清理 |
| 用户映射泄漏 | 用户离线后仍存在 | 及时清理 | 离线时清理所有映射 |

## 七、性能测试和优化效果
**测试环境**：
* 服务器：8核16G，SSD
* 操作系统：Linux
* Netty 版本：4.1.89

**测试结果**：
| **并发连接数** | **内存占用** | **CPU使用率** | **响应时间** | **说明** |
|----------|--------|--------|--------|------|
| 1万 | 200MB | 5% | < 1ms | 正常 |
| 5万 | 800MB | 15% | < 2ms | 正常 |
| 10万 | 1.5GB | 30% | < 3ms | 正常 |
| 20万 | 3GB | 60% | < 5ms | 可接受 |

**优化效果**：
* 支持连接数：从 1万 提升到 10万+
* 内存占用：每个连接数约 15KB（包含 Channel、Pipline 等）
* CPU 使用率：10万连接约 30%
* 响应时间：10万连接下 < 3ms

## 八、总结
**关键优化点**
1. Boss/Worker 线程模型：1 个 Boss + 8 个 Worker，充分利用 NIO
2. ChannelGroup 管理：使用 `DefaultChannelGroup` 实现高效广播
3. 心跳机制：10 分钟超时，自动检测和清理死连接
4. 及时清理：连接断开时清理所有相关应用
5. 内存优化：及时释放 Channel 引用，避免内存泄漏

**经验总结**
1. 线程数配置：Boss 1个，Worker 为 CPU 核数的 1-2 倍
2. 心跳超时：根据业务场景设置，IM 系统建议 5-10 分钟
3. 及时清理：连接断开时及时清理所有相关资源
4. 监控告警：监控连接数、内存占用、CPU 使用率

通过以上优化，AQChat 可支持10万+ 并发连接，为高并发 IM 系统提供基础。

在 IM 系统中，高并发连接是核心挑战。本文介绍 AQChat 如何通过 Netty 连接管理优化支持 10万+ 并发连接。

Netty 连接管理优化：如何支持 10万+ 并发连接？

数据库

Netty连接管理优化：1Boss+8Worker线程模型提升10万并发性能，ChannelGroup实现高效消息广播，10分钟心跳机制自动检测死连接，内存优化及时释放资源，8核服务器支持10万连接仅1.5GB内存占用，响应时间<3ms。

性能测试

服务器

防火墙

4核4G3M云服务器 新用户低至38元/年！

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Netty 连接管理优化：如何支持 10万+ 并发连接？-腾讯云开发者社区-腾讯云

Netty 连接管理优化：如何支持 10万+ 并发连接？

Netty 连接管理优化：如何支持 10万+ 并发连接？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐