```
话接上篇，本篇主要聚焦于该 IM 系统的架构与实现细节
```
# AQChat 即时通讯系统 - 技术深度分析文档

## 一、项目概述

**AQChat** 是一个基于 Netty 的高性能即时通讯系统，支持文本、图片、文件、音频、视频等多种消息类型，并集成了 AI 大模型能力（支持多轮对话、文本转语音、文本转图片等功能）。

### 核心特点
- **高性能**：基于 Netty 4.1.89 的 NIO 模型，支持高并发连接
- **低延迟**：使用 Protobuf 进行二进制序列化，减少网络传输开销
- **可扩展**：采用 RocketMQ 实现消息异步处理，支持分布式部署
- **AI 集成**：支持阿里百炼、Gitee AI 等多个 AI 平台，实现智能对话

---

## 二、技术架构

### 2.1 整体架构

```
┌─────────────────────────────────────────────────────────┐
│                     前端客户端 (Web/Mobile)                │
│              WebSocket 连接 (Binary Protocol)            │
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│              aq-chat-im (核心IM模块)                      │
│  ┌──────────────────────────────────────────────────┐  │
│  │  Netty Server (WebSocket)                        │  │
│  │  - MessageDecoder/Encoder (Protobuf)             │  │
│  │  - HeartBeatHandler (10分钟超时检测)              │  │
│  │  - AQChatCommandHandler (命令分发)                │  │
│  └──────────────────────────────────────────────────┘  │
│  ┌──────────────────────────────────────────────────┐  │
│  │  Handler Layer (命令处理器)                        │  │
│  │  - UserLoginCmdHandler                           │  │
│  │  - JoinRoomCmdHandler / CreateRoomCmdHandler     │  │
│  │  - SendMsgCmdHandler                             │  │
│  │  - RecallMsgCmdHandler                           │  │
│  └──────────────────────────────────────────────────┘  │
│  ┌──────────────────────────────────────────────────┐  │
│  │  Holder Layer (数据持有层)                         │  │
│  │  - GlobalChannelHolder (Channel管理)              │  │
│  │  - IUserHolder (用户信息)                         │  │
│  │  - IRoomHolder (房间信息)                         │  │
│  │  - IMessageHolder (消息去重)                      │  │
│  └──────────────────────────────────────────────────┘  │
└──────────────┬──────────────────────────┬───────────────┘
               │                          │
    ┌──────────▼──────────┐    ┌─────────▼──────────┐
    │   RocketMQ          │    │   Redis            │
    │   (消息队列)         │    │   (缓存层)          │
    └──────────┬──────────┘    └─────────┬──────────┘
               │                          │
    ┌──────────▼──────────┐    ┌─────────▼──────────┐
    │  aq-chat-service    │    │  MySQL             │
    │  (业务服务层)        │    │  (持久化存储)        │
    └─────────────────────┘    └────────────────────┘
               │
    ┌──────────▼──────────┐
    │  aq-chat-ai         │
    │  (AI服务层)          │
    │  - 阿里百炼/QW       │
    │  - Gitee AI         │
    └─────────────────────┘
```

### 2.2 模块划分

| 模块 | 职责 | 核心技术 |
|------|------|---------|
| **aq-chat-im** | IM 核心模块，负责 WebSocket 连接和消息处理 | Netty、Protobuf、RocketMQ |
| **aq-chat-service** | 业务逻辑层，处理消息持久化、用户管理等 | MyBatis-Plus、MySQL |
| **aq-chat-ai** | AI 功能模块，集成多个 AI 平台 | HTTP Client、流式处理 |
| **aq-chat-common** | 公共模块，包含工具类、常量、枚举等 | - |
| **aq-chat-framework** | 框架模块，封装中间件配置 | Spring Boot Starter |

---

## 三、核心技术实现

### 3.1 Netty WebSocket 服务器

#### 3.1.1 服务器启动

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/starter/AQChatNettyStarter.java`

**核心实现**：
```java
// 1. 创建线程组
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);  // Boss线程：1个
NioEventLoopGroup workGroup = new NioEventLoopGroup(8);  // Worker线程：8个

// 2. 配置Pipeline
ch.pipeline().addLast(new HttpServerCodec());                    // HTTP编解码
ch.pipeline().addLast(new ChunkedWriteHandler());                // 分块写入
ch.pipeline().addLast(new HttpObjectAggregator(65535));          // 聚合HTTP消息
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/"));  // WebSocket协议升级
ch.pipeline().addLast(new IdleStateHandler(0,0,10, TimeUnit.MINUTES)); // 10分钟心跳检测
ch.pipeline().addLast(messageDecoder);                           // Protobuf解码器
ch.pipeline().addLast(messageEncoder);                           // Protobuf编码器
ch.pipeline().addLast(hearBeatHandler);                          // 心跳处理器
ch.pipeline().addLast(aqChatCommandHandler);                     // 命令处理器
```

**技术要点**：
- **Boss/Worker 线程模型**：Boss 线程负责接收连接，Worker 线程处理 I/O 事件
- **心跳检测**：使用 `IdleStateHandler` 检测 10 分钟无读写，自动断开连接
- **优雅关闭**：注册 JVM 关闭钩子，确保线程池和 Netty 服务正常关闭

#### 3.1.2 消息编解码

**关键代码位置**：
- `aq-chat-im/src/main/java/com/howcode/aqchat/codec/MessageDecoder.java`
- `aq-chat-im/src/main/java/com/howcode/aqchat/codec/MessageEncoder.java`

**消息格式**：
```
+------------------+------------------+------------------+
|  消息长度(4字节)   |  消息指令(2字节)  |  消息内容(Protobuf) |
+------------------+------------------+------------------+
```

**解码流程**：
1. 读取 4 字节消息长度
2. 读取 2 字节消息指令（Command）
3. 根据指令从 `MessageRecognizer` 获取对应的 Protobuf Builder
4. 反序列化消息体

**编码流程**：
1. 将 Protobuf 消息序列化为字节数组
2. 写入 4 字节消息长度
3. 写入 2 字节消息指令
4. 写入消息体
5. 封装为 `BinaryWebSocketFrame`

**技术要点**：
- **半包/粘包处理**：通过消息长度字段判断是否接收完整消息，不完整则重置读取位置
- **消息识别器**：使用反射机制自动建立消息类型与指令的映射关系

### 3.2 命令处理机制

#### 3.2.1 命令分发

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/handler/AQChatCommandHandler.java`

**实现原理**：
- 使用 `AQChatHandlerFactory` 通过反射扫描所有 `ICmdHandler` 实现类
- 根据 `handle` 方法的第二个参数类型（Protobuf 消息类型）建立映射
- 收到消息后，根据消息类型从工厂获取对应的 Handler 执行

**设计模式**：工厂模式 + 策略模式

#### 3.2.2 核心命令处理器

| 命令 | Handler | 功能 |
|------|---------|------|
| `USER_LOGIN_CMD` | `UserLoginCmdHandler` | 用户登录，生成 userId，保存到 Redis |
| `JOIN_ROOM_CMD` | `JoinRoomCmdHandler` | 加入房间，创建 ChannelGroup |
| `CREATE_ROOM_CMD` | `CreateRoomCmdHandler` | 创建房间，生成房间号 |
| `SEND_MSG_CMD` | `SendMsgCmdHandler` | 发送消息，去重、广播、持久化 |
| `RECALL_MSG_CMD` | `RecallMsgCmdHandler` | 撤回消息 |
| `HEART_BEAT_CMD` | `HeartBeatCmdHandler` | 心跳响应 |

### 3.3 连接管理

#### 3.3.1 Channel 管理

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/holder/GlobalChannelHolder.java`

**数据结构**：
```java
// 用户ID -> Channel 映射
private final Map<String, NioSocketChannel> CHANNELS = new ConcurrentHashMap<>();

// 房间ID -> ChannelGroup 映射（在 MessageBroadcaster 中）
private final Map<String, ChannelGroup> channelGroupMap = new ConcurrentHashMap<>();
```

**核心功能**：
- **用户连接管理**：登录时建立 userId 与 Channel 的映射
- **房间管理**：每个房间维护一个 `ChannelGroup`，用于群发消息
- **离线处理**：用户断开时清理 Channel 和房间信息

#### 3.3.2 消息广播

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/message/MessageBroadcaster.java`

**实现方式**：
- 使用 Netty 的 `ChannelGroup.writeAndFlush()` 实现房间内消息广播
- 支持单播（通过 userId 获取 Channel）和广播（通过 roomId 获取 ChannelGroup）

### 3.4 消息去重机制

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/holder/impl/AQMessageHolder.java`

**实现原理**：
- 使用 Redis Set 存储房间内已发送的消息 ID
- 发送消息前检查 `SISMEMBER`，防止重复处理
- 消息 ID 由客户端生成（UUID），服务端校验

**Redis Key 格式**：`AQChat:room:message:{roomId}`

### 3.5 RocketMQ 消息队列

#### 3.5.1 消息主题

| Topic | 用途 | 消费者 |
|-------|------|--------|
| `SEND_MESSAGE_TOPIC` | 消息广播 | `MessageBroadcastReceiver` |
| `STORE_MESSAGE_TOPIC` | 消息持久化 | `StoreMessageReceiver` |
| `JOIN_ROOM_TOPIC` | 加入房间通知 | `JoinRoomReceiver` |
| `LEAVE_ROOM_TOPIC` | 离开房间通知 | `LeaveRoomReceiver` |
| `OFFLINE_MESSAGE_TOPIC` | 用户离线通知 | `OfflineMessageReceiver` |
| `AI_HELPER_TOPIC` | AI 助手消息 | `AIHelperReceiver` |
| `RECALL_MESSAGE_TOPIC` | 撤回消息 | `RecallMessageReceiver` |

#### 3.5.2 消息发送流程

**发送消息流程**：
```
用户发送消息
    ↓
SendMsgCmdHandler 处理
    ↓
1. 消息去重检查
2. 发送到 SEND_MESSAGE_TOPIC (异步广播)
3. 发送到 STORE_MESSAGE_TOPIC (异步持久化)
4. 如果@AI，发送到 AI_HELPER_TOPIC
    ↓
返回 SendMsgAck
```

**技术要点**：
- **异步处理**：消息广播和持久化通过 MQ 异步处理，提高响应速度
- **延迟消息**：用户离线时发送延迟消息（30 分钟）用于清理遗留数据
- **消息模型**：使用 `CLUSTERING` 模式，支持负载均衡

### 3.6 Redis 缓存设计

#### 3.6.1 数据结构

| Key 前缀 | 数据结构 | 用途 | TTL |
|---------|---------|------|-----|
| `AQChat:userInfo:{userId}` | String | 用户信息 | 30 分钟 |
| `AQChat:room:roomNo:{roomNo}` | String | 房间号 -> 房间ID 映射 | 永久 |
| `AQChat:room:roomData:{roomId}` | Hash | 房间信息、成员列表 | 永久 |
| `AQChat:room:message:{roomId}` | Set | 消息ID去重 | 永久 |
| `AQChat:aliOssSts` | String | OSS 临时凭证 | 1 小时 |

#### 3.6.2 房间信息存储

**Hash 结构**：
```
Key: AQChat:room:roomData:{roomId}
Field: info -> RoomInfoDto (JSON)
Field: member -> Map<userId, UserGlobalInfoDto> (JSON)
```

**技术要点**：
- 使用 Hash 结构存储房间的多个属性，减少 Redis 键数量
- 成员列表使用 Map 结构，方便添加/删除成员

### 3.7 AI 功能集成

#### 3.7.1 AI 服务架构

**关键代码位置**：`aq-chat-ai/src/main/java/com/howcode/aqchat/ai/`

**支持的 AI 平台**：
1. **阿里百炼（通义千问）**：`QWAiService`
2. **Gitee AI**：`GiteeAIService`
   - 文本对话：`llama3-70b-chinese-chat`、`Qwen2-72B-Instruct`
   - 文本转图片：`stable-diffusion-3-medium`
   - 文本转语音：`ChatTTS`

#### 3.7.2 流式响应处理

**实现原理**：
- 使用 `Consumer<AIResult>` 回调函数处理流式数据
- AI 返回的每个数据块通过回调函数发送到前端
- 前端通过 `STREAM_MSG_NOTIFY` 消息类型实时显示

**关键代码**：
```java
// 流式调用
aiService.streamCallWithMessage(userMsg, result -> {
    // 构造流式消息
    AIMessageDto aiMessageDto = new AIMessageDto();
    aiMessageDto.setContent(result.getContent());
    aiMessageDto.setStatus(result.getStatus()); // WAIT/END/FAIL
    // 广播到房间
    globalChannelHolder.sendBroadcastAIMessage(aiMessageDto, aiId);
});
```

#### 3.7.3 @AI 功能

**实现原理**：
- 使用责任链模式处理 @ 消息
- 责任链：`XQHandler` -> `XTHandler` -> `XMHandler` -> `XVHandler` -> `MultipleRoundsHandler`
- 每个 Handler 检查消息扩展字段（ext）是否包含对应的 @ 标识
- 匹配则触发对应的 AI 服务，否则传递给下一个 Handler

### 3.8 文件上传（OSS）

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/handler/impl/GetStsCmdHandler.java`

**实现流程**：
1. 客户端请求获取 OSS 临时凭证（STS）
2. 服务端调用阿里云 STS 服务获取临时 AccessKey
3. 返回临时凭证给客户端
4. 客户端直接上传文件到 OSS
5. 上传成功后，将 OSS URL 作为消息内容发送

**技术要点**：
- **临时凭证**：使用 STS 临时访问凭证，避免暴露主账号密钥
- **凭证缓存**：Redis 缓存 1 小时，减少 STS 调用次数
- **路径规划**：按消息类型和时间组织上传路径

### 3.9 消息撤回

**实现流程**：
1. 客户端发送 `RECALL_MSG_CMD`
2. 服务端验证用户权限（只能撤回自己的消息）
3. 发送到 `RECALL_MESSAGE_TOPIC`
4. MQ 消费者更新数据库消息状态为 0（不显示）
5. 广播 `RECALL_MSG_NOTIFY` 到房间内所有用户

**数据库更新**：`UPDATE aq_message SET status = 0 WHERE message_id = ?`

### 3.10 心跳机制

**关键代码位置**：`aq-chat-im/src/main/java/com/howcode/aqchat/handler/HearBeatHandler.java`

**实现原理**：
- 使用 `IdleStateHandler` 检测 10 分钟无读写事件
- 触发 `ALL_IDLE` 事件时，发送离线通知并关闭连接
- 客户端定期发送心跳包（`HEART_BEAT_CMD`），重置空闲时间

---

## 四、技术亮点

### 4.1 高性能设计

1. **NIO 模型**：Netty 的 Reactor 模式，单线程处理大量连接
2. **二进制协议**：Protobuf 序列化，比 JSON 体积小 30-50%
3. **异步处理**：消息广播和持久化通过 MQ 异步处理，响应时间 < 10ms
4. **连接复用**：WebSocket 长连接，避免频繁建立连接

### 4.2 高可用设计

1. **消息去重**：Redis Set 防止重复处理，支持幂等性
2. **优雅降级**：AI 服务异常时不影响正常聊天功能
3. **连接保活**：心跳机制检测死连接，及时清理资源
4. **数据持久化**：消息异步持久化到 MySQL，保证数据不丢失

### 4.3 可扩展设计

1. **模块化架构**：5 个独立模块，便于维护和扩展
2. **命令模式**：新增命令只需实现 `ICmdHandler` 接口
3. **责任链模式**：AI 处理使用责任链，易于扩展新的 AI 服务
4. **MQ 解耦**：业务逻辑与消息处理解耦，支持水平扩展

### 4.4 安全性设计

1. **XSS 防护**：消息内容过滤 HTML 标签（`SafeUtil`）
2. **临时凭证**：文件上传使用 STS 临时凭证，避免密钥泄露
3. **权限校验**：每个命令处理器都验证用户登录状态和房间权限


---

## 五、补充说明

### 5.1 项目规模
- **代码量**：约 5000+ 行 Java 代码
- **模块数**：5 个独立模块
- **命令类型**：30+ 种消息命令
- **AI 平台**：2 个（阿里百炼、Gitee AI）

### 5.2 可扩展方向
1. **分布式部署**：实现 Channel 路由，支持多机部署
2. **消息推送**：集成推送服务（极光、个推），支持离线消息推送
3. **音视频通话**：集成 WebRTC，支持音视频通话
4. **消息加密**：使用 TLS/SSL 加密消息传输
5. **监控告警**：集成 Prometheus + Grafana，监控系统指标


AQChat 是一个基于 Netty 的高性能即时通讯系统，支持文本、图片、文件、音频、视频等多种消息类型，并集成了 AI 大模型能力（支持多轮对话、文本转语音、文本转图片等功能）。

AQChat 即时通讯系统 - 技术深度分析文档

后端

网络与通信

架构设计

AQChat是一款基于Netty的高性能即时通讯系统，支持文本、图片、文件、音视频等多种消息类型，集成AI大模型能力。采用Netty+NIO架构，Protobuf序列化，RocketMQ异步处理，实现高并发低延迟。系统具备消息去重、心跳检测、文件上传等核心功能，支持分布式部署，适用于企业级即时通讯场景。

消息队列

负载均衡

数据结构

服务器

数据库

中间件

路由

Redis

MySQL

JSON

Java

云数据库采购特惠，高性能、高可靠、高安全，超值优惠等你享！


AI驱动 智领未来

message-queue-catalog

tencentdb-catalog

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

2万分钟实时音视频通用时长新用户9.9元


trtc

5位重磅专家聚焦 AI 时代开发者核心成长需求，带你深入解锁全栈进化、人机协作关键路径！点击报名到场：<a href="https://cloud.tencent.com/developer/salon/salon-2403/signup" target="_blank">https://cloud.tencent.com/developer/salon/salon-2403/signup</a><br/> 
 <img src="https://qcloudimg.tencent-cloud.cn/raw/487c82639d2338df3ef66dfb1450601a.png"/>

腾讯技术创作特训营「年度论坛」正在报名中！

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

AQChat 即时通讯系统 - 技术深度分析文档-腾讯云开发者社区-腾讯云

AQChat 即时通讯系统 - 技术深度分析文档

AQChat 即时通讯系统 - 技术深度分析文档

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐