🎯 核心挑战 — 5大难题
挑战1：消息必达 — 消息不能丢，丢了就是事故
IM系统最基本的要求：消息不能丢。你发给朋友的消息，朋友必须收到。网络断了重连后，丢失的消息要补上。
⚠️ 为什么消息会丢？：网络不稳定（WiFi切4G、进电梯信号消失）、服务端重启、客户端崩溃——任何环节出问题都可能导致消息丢失。就像你寄挂号信，邮递员可能在路上摔倒信丢了，但你必须确保信最终送到。
类比理解：挂号信必须送到，收不到要反复投递。普通信丢了就算了，但挂号信丢了邮局要负责——IM消息就是挂号信。
消息投递模型
发送方
→
服务端存储
→
推送
→
接收方
→
ACK
→
标记已送达
未收到ACK？→ 超时重推 → 直到收到ACK或超过重试次数
核心思路：先存后推。消息到达服务端后，先存到数据库（持久化），然后再推给接收方。如果推送失败，消息不会丢——因为已经存了，重推就行。
✅ 消息必达三原则：① 服务端先持久化再推送（不丢） ② 客户端必须回ACK（确认） ③ 未ACK超时重推（补上）
消息投递状态机
发送中
→
已送达(服务端)
→
已送达(客户端)
→
已读
消息投递状态从"发送中"一步步推进到"已读"，每一步都有确认机制，确保不遗漏。
未ACK重推机制代码实现
// 消息投递服务 — 核心投递逻辑
public class MessageDeliveryService {
// 消息投递状态枚举
public enum DeliveryStatus {
SENDING, // 发送中
SERVER_RECEIVED, // 服务端已接收
DELIVERED, // 已送达客户端
READ // 已读
}
@Autowired
private MessageRepository messageRepo;
@Autowired
private WebSocketSessionManager sessionManager;
@Autowired
private RedisTemplate<String, String> redisTemplate;
// 最大重试次数
private static final int MAX_RETRY = 5;
// 重试间隔（毫秒）: 1s, 2s, 4s, 8s, 16s 指数退避
private static final long[] RETRY_INTERVALS = {
1000, 2000, 4000, 8000, 16000
};
/**
* 发送消息 — 核心投递流程
* 1. 先存DB（持久化，保证不丢）
* 2. 再推WebSocket
* 3. 等ACK，未ACK则重推
*/
public void sendMessage(Message msg) {
// 第1步：持久化消息到数据库（先存后推！）
msg.setStatus(DeliveryStatus.SERVER_RECEIVED);
msg.setRetryCount(0);
messageRepo.save(msg);
// 第2步：推送到接收方
pushMessage(msg);
}
/**
* 推送消息到接收方的WebSocket连接
*/
private void pushMessage(Message msg) {
String userId = msg.getReceiverId();
WebSocketSession session = sessionManager.getSession(userId);
if (session != null && session.isOpen()) {
// 对方在线，通过WebSocket推送
try {
String json = JsonUtil.toJson(msg);
session.sendMessage(new TextMessage(json));
// 设置ACK超时检测
scheduleAckTimeout(msg);
} catch (Exception e) {
// 推送失败，进入重试队列
scheduleRetry(msg);
}
} else {
// 对方不在线，存入离线消息队列
storeOfflineMessage(msg);
}
}
/**
* 处理ACK — 客户端确认收到消息
*/
public void handleAck(String messageId, String userId) {
Message msg = messageRepo.findById(messageId);
if (msg != null) {
msg.setStatus(DeliveryStatus.DELIVERED);
messageRepo.save(msg);
// 取消重试任务
cancelRetryTask(messageId);
// 通知发送方：消息已送达
notifySender(msg.getSenderId(), messageId, "delivered");
}
}
/**
* ACK超时 — 未收到确认，触发重试
*/
private void scheduleAckTimeout(Message msg) {
// 5秒内未收到ACK，触发重试
ScheduledFuture<?> future = scheduler.schedule(() -> {
Message latest = messageRepo.findById(msg.getId());
if (latest != null && latest.getStatus() != DeliveryStatus.DELIVERED) {
scheduleRetry(latest);
}
}, 5, TimeUnit.SECONDS);
retryTasks.put(msg.getId(), future);
}
/**
* 重试推送 — 指数退避策略
*/
private void scheduleRetry(Message msg) {
int retryCount = msg.getRetryCount();
if (retryCount >= MAX_RETRY) {
// 超过最大重试次数，标记为投递失败，转离线存储
storeOfflineMessage(msg);
return;
}
msg.setRetryCount(retryCount + 1);
messageRepo.save(msg);
// 指数退避：1s → 2s → 4s → 8s → 16s
long delay = RETRY_INTERVALS[Math.min(retryCount, RETRY_INTERVALS.length - 1)];
scheduler.schedule(() -> pushMessage(msg), delay, TimeUnit.MILLISECONDS);
}
/**
* 存入离线消息队列（Redis List）
*/
private void storeOfflineMessage(Message msg) {
String key = "offline:" + msg.getReceiverId();
redisTemplate.opsForList().rightPush(key, JsonUtil.toJson(msg));
// 设置7天过期
redisTemplate.expire(key, 7, TimeUnit.DAYS);
}
}
🔑 关键点：先存后推是消息必达的核心。就像挂号信——邮局先登记你的信（持久化），然后再派送。即使派送途中出了问题，信还在邮局，可以重新派送。如果直接推不存，就像口头传话——中间人忘了，话就丢了。
挑战2：已读未读 — 对方看了没有？
你发完消息后，最关心的是什么？"对方看了没有！"这就是已读未读的核心需求。
类比理解：已读回执就像挂号信的签收回执。你寄出一封挂号信，邮局会给你一张回执，告诉你对方已经签收了。微信的"已读"就是这种签收回执。
💡 单聊已读 vs 群聊已读：单聊简单——一对一，更新一条记录就行。群聊复杂——500人的群，谁看了谁没看？这就需要仔细设计了。
单聊已读：点对点更新
单聊已读很简单，只需要维护一个"已读位置"——记录对方读到了第几条消息。
// 已读状态表设计
// 表名: read_status
// +-----------+----------+------+
// | session_id| user_id | last_read_msg_id |
// +-----------+----------+------+
// | s_001 | user_B | msg_105 |
// +-----------+----------+------+
// 含义：在会话s_001中，用户B已读的最后一条消息是msg_105
public class ReadStatusService {
@Autowired
private ReadStatusRepository readStatusRepo;
@Autowired
private WebSocketSessionManager sessionManager;
/**
* 标记已读 — 用户打开聊天界面时调用
* @param sessionId 会话ID
* @param userId 当前用户ID
* @param lastReadMsgId 已读的最后一条消息ID
*/
public void markAsRead(String sessionId, String userId, String lastReadMsgId) {
// 更新已读位置
ReadStatus status = readStatusRepo.findBySessionAndUser(sessionId, userId);
if (status == null) {
status = new ReadStatus();
status.setSessionId(sessionId);
status.setUserId(userId);
}
status.setLastReadMsgId(lastReadMsgId);
status.setReadTime(new Date());
readStatusRepo.save(status);
// 通知对方：消息已读
String otherUserId = getOtherUserId(sessionId, userId);
ReadNotification notification = new ReadNotification();
notification.setSessionId(sessionId);
notification.setLastReadMsgId(lastReadMsgId);
notification.setReadBy(userId);
sessionManager.sendToUser(otherUserId,
JsonUtil.toJson(notification));
}
/**
* 获取未读消息数
*/
public int getUnreadCount(String sessionId, String userId) {
ReadStatus status = readStatusRepo.findBySessionAndUser(sessionId, userId);
if (status == null) {
// 从未读过，返回该会话所有消息数
return messageRepo.countBySession(sessionId);
}
// 统计lastReadMsgId之后的消息数
return messageRepo.countAfterId(sessionId, status.getLastReadMsgId());
}
}
群聊已读：计数器方案 vs 明细方案
群聊已读就复杂了——500人的群，每个人是否已读都要记录。两种方案各有取舍：
| 对比维度 |
计数器方案 |
明细方案 |
| 原理 |
每条消息维护一个"已读人数"计数器 |
每条消息记录"谁已读了"的明细列表 |
| 存储量 |
极小：1条消息 = 1个数字 |
较大：1条消息 × 500人 = 500条记录 |
| 查询已读人数 |
O(1)，直接读计数器 |
O(N)，需要count查询 |
| 查询谁已读/未读 |
做不到，只有数字 |
O(N)，可以精确列出 |
| 适用场景 |
只关心"几个人已读"的场景 |
需要知道"谁读了谁没读"的场景 |
| 典型应用 |
微信群聊（只显示已读人数） |
钉钉群聊（显示谁读了谁没读） |
| 性能 |
高，计数器操作极快 |
低，写入量大，查询慢 |
// ========== 方案1：计数器方案 ==========
// Redis Key设计: group:read:count:{groupId}:{msgId} → 已读人数
// Redis Key设计: group:read:total:{groupId}:{msgId} → 总人数
public class GroupReadCounterService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private StringRedisTemplate stringRedis;
/**
* 群消息已读 — 计数器方案
* 每个用户已读一次，计数器+1
*/
public void markGroupRead(String groupId, String msgId, String userId) {
// 用Set去重：同一个用户对同一条消息只能算一次已读
String setKey = "group:read:set:" + groupId + ":" + msgId;
Long added = stringRedis.opsForSet().add(setKey, userId);
if (added > 0) {
// 第一次标记已读，计数器+1
String countKey = "group:read:count:" + groupId + ":" + msgId;
redisTemplate.opsForValue().increment(countKey);
}
}
/**
* 获取群消息已读人数
*/
public int getReadCount(String groupId, String msgId) {
String countKey = "group:read:count:" + groupId + ":" + msgId;
String count = redisTemplate.opsForValue().get(countKey);
return count == null ? 0 : Integer.parseInt(count);
}
/**
* 获取未读人数 = 群总人数 - 已读人数
*/
public int getUnreadCount(String groupId, String msgId, int totalMembers) {
return totalMembers - getReadCount(groupId, msgId);
}
}
// ========== 方案2：明细方案 ==========
// 表名: group_read_detail
// +---------+--------+--------+-----------+
// | group_id| msg_id | user_id| read_time |
// +---------+--------+--------+-----------+
// | g_001 | msg_10 | user_A | 10:30:00 |
// | g_001 | msg_10 | user_B | 10:31:00 |
// +---------+--------+--------+-----------+
public class GroupReadDetailService {
@Autowired
private GroupReadDetailRepository readDetailRepo;
/**
* 标记群消息已读 — 明细方案
*/
public void markGroupRead(String groupId, String msgId, String userId) {
// 幂等：已存在则不重复插入
if (!readDetailRepo.existsByGroupAndMsgAndUser(groupId, msgId, userId)) {
GroupReadDetail detail = new GroupReadDetail();
detail.setGroupId(groupId);
detail.setMsgId(msgId);
detail.setUserId(userId);
detail.setReadTime(new Date());
readDetailRepo.save(detail);
}
}
/**
* 查询谁已读了（钉钉场景：显示已读人列表）
*/
public List<String> getReadUsers(String groupId, String msgId) {
return readDetailRepo.findReadUserIds(groupId, msgId);
}
/**
* 查询谁未读（钉钉场景：显示未读人列表，方便@提醒）
*/
public List<String> getUnreadUsers(String groupId, String msgId) {
List<String> allMembers = groupMemberRepo.findMemberIds(groupId);
List<String> readUsers = getReadUsers(groupId, msgId);
allMembers.removeAll(readUsers);
return allMembers;
}
}
💡 实战建议：大多数场景用计数器方案就够了（存储小、性能高）。只有像钉钉这种需要精确显示"谁读了谁没读"的场景，才用明细方案。也可以混合使用：Redis计数器做实时展示，异步写入DB明细做后续分析。
挑战3：群聊扩散 — 1条消息推给500人
群聊最大的挑战：1条消息要推给群里的所有人。500人的群，发1条消息就要投递500次。万人大群呢？10万人大群呢？这就是"群聊扩散"问题。
类比理解：班主任通知全班——不需要跟每人单独说，发到班群里，所有人自己来看就行。但如果是"写扩散"（给每人单独发），就是老师给50个学生分别打电话，太累了。
写扩散 vs 读扩散
| 对比维度 |
写扩散（推模式） |
读扩散（拉模式） |
| 原理 |
发消息时，写入每个成员的收件箱 |
发消息时，只写入群信箱；读消息时，从群信箱拉取 |
| 写入量 |
大：1条消息 × N人 = N次写入 |
小：1条消息 = 1次写入 |
| 读取量 |
小：直接读自己收件箱 |
大：每次打开群聊都要查群信箱 |
| 写延迟 |
高：要写N份，大群写完要几秒 |
低：只写1份，瞬间完成 |
| 读延迟 |
低：直接读自己的 |
较高：需要查群信箱+过滤 |
| 适用场景 |
小群（<200人），读多写少 |
大群（>200人），写多读少 |
| 典型应用 |
微信小群聊 |
微博超话、B站直播间 |
写扩散
消息
→
写N份
→
用户1收件箱
用户2收件箱
...用户N收件箱
读扩散
消息
→
写1份到群信箱
→
用户打开群聊时拉取
混合方案：小群写扩散、大群读扩散
✅ 实战最优解：人数 ≤ 200 用写扩散（每人收件箱直接有消息，读起来快），人数 > 200 用读扩散（只写群信箱，读时再拉）。这样兼顾了小群的体验和大群的性能。
public class GroupMessageDispatchService {
// 写扩散阈值：群成员数 <= 200 用写扩散
private static final int WRITE_FANOUT_THRESHOLD = 200;
@Autowired
private GroupMemberRepository groupMemberRepo;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private RocketMQTemplate mqTemplate;
@Autowired
private MessageDeliveryService deliveryService;
/**
* 群消息扩散 — 混合方案
* 1. 消息先存到群消息表
* 2. 根据群大小选择扩散策略
*/
public void dispatchGroupMessage(Message msg, String groupId) {
// 第1步：消息存入群消息表（所有方案都需要）
groupMessageRepo.save(msg);
// 第2步：获取群成员数
int memberCount = groupMemberRepo.countByGroup(groupId);
if (memberCount <= WRITE_FANOUT_THRESHOLD) {
// 小群：写扩散 — 给每个在线成员推送
writeFanout(msg, groupId);
} else {
// 大群：读扩散 — 只更新群信箱时间线
readFanout(msg, groupId);
}
}
/**
* 写扩散：给每个群成员推送消息
*/
private void writeFanout(Message msg, String groupId) {
List<String> memberIds = groupMemberRepo.findMemberIds(groupId);
// 异步批量推送，不阻塞发送方
for (String memberId : memberIds) {
if (!memberId.equals(msg.getSenderId())) {
// 排除发送者自己
Message copy = msg.copyForReceiver(memberId);
// 写入每个成员的收件箱Timeline（Redis Sorted Set）
String inboxKey = "inbox:" + memberId;
redisTemplate.opsForZSet().add(
inboxKey,
copy.getId(),
msg.getTimestamp()
);
// 如果在线，实时推送
deliveryService.pushMessage(copy);
}
}
}
/**
* 读扩散：只更新群信箱时间线
* 用户打开群聊时，再从群信箱拉取
*/
private void readFanout(Message msg, String groupId) {
// 只写入群信箱时间线
String groupInboxKey = "group:inbox:" + groupId;
redisTemplate.opsForZSet().add(
groupInboxKey,
msg.getId(),
msg.getTimestamp()
);
// 通知在线成员：有新群消息（只通知，不推送内容）
List<String> onlineMembers = sessionManager.getOnlineMembers(groupId);
for (String memberId : onlineMembers) {
if (!memberId.equals(msg.getSenderId())) {
GroupNotification notify = new GroupNotification();
notify.setGroupId(groupId);
notify.setMsgId(msg.getId());
notify.setType("NEW_MSG");
sessionManager.sendToUser(memberId, JsonUtil.toJson(notify));
}
}
}
}
挑战4：长连接管理 — WebSocket连接建立、心跳、断线重连
IM系统的基础是WebSocket长连接。就像打电话——拨号建立连接、通话中保持连接、沉默太久问一句"还在吗"（心跳）、信号不好断了自动重拨（断线重连）。
WebSocket连接生命周期
客户端发起连接
→
握手(HTTP→WS升级)
→
连接建立
连接建立
→
心跳保活(每30s)
→
连接正常
连接正常
→
心跳超时(3次无响应)
→
判定断开
判定断开
→
自动重连
→
连接恢复
💡 为什么需要心跳？：网络中存在"半开连接"——客户端断网了，但服务端不知道，还以为连接在。就像打电话时对方手机没电了，你还在说话，其实对方已经听不到了。心跳机制就是每隔几秒问一句"还在吗"，对方不回答就判定断开。
心跳机制 + 断线重连代码
// ========== 服务端心跳检测 ==========
@ServerEndpoint(value = "/ws/im/{userId}")
public class ImWebSocketServer {
// 心跳超时时间：60秒无心跳判定断开
private static final long HEARTBEAT_TIMEOUT = 60_000;
// 存储所有在线连接: userId → Session
private static final ConcurrentHashMap<String, Session> onlineSessions
= new ConcurrentHashMap<>();
// 存储最后心跳时间: userId → 最后心跳时间
private static final ConcurrentHashMap<String, Long> lastHeartbeat
= new ConcurrentHashMap<>();
/**
* 连接建立
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
onlineSessions.put(userId, session);
lastHeartbeat.put(userId, System.currentTimeMillis());
log.info("用户上线: {}", userId);
}
/**
* 收到消息
*/
@OnMessage
public void onMessage(String message, Session session) {
JsonNode node = ObjectMapper.readTree(message);
String type = node.get("type").asText();
if ("heartbeat".equals(type)) {
// 收到心跳，更新最后心跳时间
String userId = getUserIdFromSession(session);
lastHeartbeat.put(userId, System.currentTimeMillis());
// 回复心跳
sendToSession(session, "{\"type\":\"heartbeat_ack\"}");
} else if ("message".equals(type)) {
// 处理业务消息
messageService.handleMessage(node);
}
}
/**
* 连接关闭
*/
@OnClose
public void onClose(Session session) {
String userId = getUserIdFromSession(session);
onlineSessions.remove(userId);
lastHeartbeat.remove(userId);
log.info("用户下线: {}", userId);
}
/**
* 心跳检测定时任务 — 每隔30秒扫描一次
* 清理超过60秒没有心跳的连接
*/
public static void checkHeartbeat() {
long now = System.currentTimeMillis();
for (Map.Entry<String, Long> entry : lastHeartbeat.entrySet()) {
if (now - entry.getValue() > HEARTBEAT_TIMEOUT) {
String userId = entry.getKey();
Session session = onlineSessions.get(userId);
if (session != null) {
// 关闭超时连接
try {
session.close();
} catch (Exception e) {
log.warn("关闭超时连接失败: {}", userId);
}
}
onlineSessions.remove(userId);
lastHeartbeat.remove(userId);
log.info("心跳超时，清理连接: {}", userId);
}
}
}
}
// ========== 客户端断线重连（JavaScript）==========
class IMClient {
constructor(userId) {
this.userId = userId;
this.ws = null;
this.heartbeatTimer = null;
this.reconnectTimer = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 10;
// 重连间隔：1s, 2s, 4s, 8s, 16s, 32s, 60s, 60s...
this.reconnectIntervals = [1000, 2000, 4000, 8000,
16000, 32000, 60000];
this.connect();
}
// 建立连接
connect() {
this.ws = new WebSocket(`ws://server/ws/im/${this.userId}`);
this.ws.onopen = () => {
console.log('WebSocket连接成功');
this.reconnectAttempts = 0; // 重置重连次数
this.startHeartbeat(); // 启动心跳
this.pullOfflineMessages(); // 拉取离线消息
};
this.ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'heartbeat_ack') {
// 心跳回复，连接正常
return;
}
if (data.type === 'message') {
this.handleMessage(data);
// 发送ACK确认
this.sendAck(data.id);
}
};
this.ws.onclose = () => {
console.log('WebSocket连接关闭');
this.stopHeartbeat();
this.scheduleReconnect(); // 自动重连
};
this.ws.onerror = (err) => {
console.error('WebSocket错误', err);
};
}
// 心跳保活 — 每30秒发一次
startHeartbeat() {
this.heartbeatTimer = setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({type: 'heartbeat'}));
}
}, 30000);
}
stopHeartbeat() {
clearInterval(this.heartbeatTimer);
}
// 断线重连 — 指数退避
scheduleReconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.log('超过最大重连次数，停止重连');
return;
}
const delay = this.reconnectIntervals[
Math.min(this.reconnectAttempts, this.reconnectIntervals.length - 1)
];
console.log(`${delay}ms后尝试第${this.reconnectAttempts + 1}次重连`);
this.reconnectTimer = setTimeout(() => {
this.reconnectAttempts++;
this.connect();
}, delay);
}
// 拉取离线消息
pullOfflineMessages() {
fetch(`/api/messages/offline?userId=${this.userId}`)
.then(res => res.json())
.then(messages => {
messages.forEach(msg => this.handleMessage(msg));
});
}
}
挑战5：离线消息 — 不在线时的消息怎么存？
用户不在线时，发给他的消息怎么处理？存起来，上线后再取。
类比理解：你出差了，快递员来送包裹你不在家。快递员不会把包裹扔掉，而是放在快递柜里（离线存储）。等你回来，拿取件码去快递柜取就行了。
发送方发消息
→
服务端发现接收方不在线
→
存入Redis List(离线队列)
接收方上线
→
拉取离线消息
→
展示给用户
→
清空离线队列
public class OfflineMessageService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private MessageRepository messageRepo;
// 离线消息Key前缀: offline:{userId}
private static final String OFFLINE_KEY_PREFIX = "offline:";
// 离线消息最大保存条数
private static final int MAX_OFFLINE_MSG = 1000;
// 离线消息过期天数
private static final int EXPIRE_DAYS = 7;
/**
* 存储离线消息
* 用户不在线时，消息存入Redis List
*/
public void storeOffline(Message msg) {
String key = OFFLINE_KEY_PREFIX + msg.getReceiverId();
String json = JsonUtil.toJson(msg);
// 右入队（先进先出）
redisTemplate.opsForList().rightPush(key, json);
// 限制离线消息数量，防止堆积
Long size = redisTemplate.opsForList().size(key);
if (size != null && size > MAX_OFFLINE_MSG) {
// 移除最旧的消息
redisTemplate.opsForList().leftPop(key);
}
// 设置过期时间
redisTemplate.expire(key, EXPIRE_DAYS, TimeUnit.DAYS);
}
/**
* 拉取离线消息
* 用户上线后调用，一次性取出所有离线消息
*/
public List<Message> pullOfflineMessages(String userId) {
String key = OFFLINE_KEY_PREFIX + userId;
// 取出所有离线消息
List<String> jsonList = redisTemplate.opsForList().range(key, 0, -1);
if (jsonList == null || jsonList.isEmpty()) {
return Collections.emptyList();
}
// 反序列化
List<Message> messages = jsonList.stream()
.map(json -> JsonUtil.fromJson(json, Message.class))
.collect(Collectors.toList());
// 取完后清空离线队列
redisTemplate.delete(key);
return messages;
}
/**
* 获取离线消息数量（用于显示角标）
*/
public long getOfflineCount(String userId) {
String key = OFFLINE_KEY_PREFIX + userId;
Long size = redisTemplate.opsForList().size(key);
return size == null ? 0 : size;
}
}
🔄 关键流程 — 完整代码实现
流程1：发消息完整流程（Spring Boot + WebSocket + Netty）
客户端A发送
→
WS网关接收
→
消息持久化
→
消息路由
→
推送/离线存储
→
等待ACK
// ========== 1. Netty WebSocket服务端 ==========
public class NettyWebSocketServer {
private final int port;
public NettyWebSocketServer(int port) {
this.port = port;
}
public void start() throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// HTTP编解码（用于WebSocket握手）
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
// WebSocket协议处理
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
// 空闲检测（心跳）
pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
// 业务处理器
pipeline.addLast(new ImMessageHandler());
}
});
Channel channel = bootstrap.bind(port).sync().channel();
log.info("WebSocket服务器启动，端口: {}", port);
channel.closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
// ========== 2. 消息业务处理器 ==========
public class ImMessageHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (frame instanceof TextWebSocketFrame) {
String text = ((TextWebSocketFrame) frame).text();
JsonNode node = JsonUtil.parse(text);
String type = node.get("type").asText();
switch (type) {
case "message":
handleChatMessage(ctx, node);
break;
case "heartbeat":
handleHeartbeat(ctx);
break;
case "ack":
handleAck(node);
break;
case "read":
handleRead(node);
break;
}
}
}
/**
* 处理聊天消息
*/
private void handleChatMessage(ChannelHandlerContext ctx, JsonNode node) {
Message msg = JsonUtil.fromJson(node.get("data"), Message.class);
// 第1步：消息持久化（先存后推）
messageService.saveMessage(msg);
// 第2步：发送方确认（消息已到达服务端）
sendToChannel(ctx, JsonUtil.toJson(AckResponse.serverReceived(msg.getId())));
// 第3步：消息路由 + 投递
if (msg.isGroupMessage()) {
// 群消息走扩散逻辑
groupDispatchService.dispatchGroupMessage(msg, msg.getGroupId());
} else {
// 单聊走路由投递
routeService.routeMessage(msg);
}
}
/**
* 空闲检测 — 心跳超时处理
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
// 60秒没有读到数据，关闭连接
log.warn("连接空闲超时，关闭: {}", ctx.channel());
ctx.close();
}
}
}
// ========== 3. 消息发送API ==========
@RestController
@RequestMapping("/api/messages")
public class MessageController {
@Autowired
private MessageService messageService;
@Autowired
private OfflineMessageService offlineService;
/**
* 发送消息
*/
@PostMapping("/send")
public Result<String> sendMessage(@RequestBody SendMessageRequest req) {
Message msg = new Message();
msg.setId(IdGenerator.nextId());
msg.setSenderId(req.getSenderId());
msg.setReceiverId(req.getReceiverId());
msg.setContent(req.getContent());
msg.setTimestamp(System.currentTimeMillis());
messageService.sendMessage(msg);
return Result.success(msg.getId());
}
/**
* 拉取离线消息
*/
@GetMapping("/offline")
public Result<List<Message>> pullOffline(@RequestParam String userId) {
List<Message> messages = offlineService.pullOfflineMessages(userId);
return Result.success(messages);
}
}
流程2：群消息扩散完整代码
// 群消息发送 — 混合扩散策略完整流程
@Service
public class GroupMessageService {
private static final int WRITE_FANOUT_THRESHOLD = 200;
private static final int BATCH_SIZE = 50;
@Autowired
private GroupMemberRepository groupMemberRepo;
@Autowired
private RocketMQTemplate mqTemplate;
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 发送群消息 — 入口
*/
@Transactional
public String sendGroupMessage(String groupId, String senderId, String content) {
// 1. 生成消息ID（雪花算法，保证有序且唯一）
String msgId = SnowflakeIdGenerator.nextId();
// 2. 消息持久化到群消息表
GroupMessage groupMsg = new GroupMessage();
groupMsg.setId(msgId);
groupMsg.setGroupId(groupId);
groupMsg.setSenderId(senderId);
groupMsg.setContent(content);
groupMsg.setTimestamp(System.currentTimeMillis());
groupMessageRepo.save(groupMsg);
// 3. 更新群消息时间线（读扩散用）
String groupTimelineKey = "group:timeline:" + groupId;
redisTemplate.opsForZSet().add(groupTimelineKey, msgId, groupMsg.getTimestamp());
// 4. 异步扩散
int memberCount = groupMemberRepo.countByGroup(groupId);
if (memberCount <= WRITE_FANOUT_THRESHOLD) {
// 小群：写扩散（异步MQ，分批写入）
List<String> memberIds = groupMemberRepo.findMemberIds(groupId);
// 分批发送MQ消息，避免单条消息太大
Lists.partition(memberIds, BATCH_SIZE).forEach(batch -> {
FanoutMessage fanoutMsg = new FanoutMessage();
fanoutMsg.setMsgId(msgId);
fanoutMsg.setGroupId(groupId);
fanoutMsg.setContent(content);
fanoutMsg.setSenderId(senderId);
fanoutMsg.setTargetUserIds(batch);
mqTemplate.syncSend("group-fanout", fanoutMsg);
});
} else {
// 大群：读扩散 — 只更新群信箱 + 通知在线成员
List<String> onlineMembers = sessionManager.getOnlineMembers(groupId);
for (String memberId : onlineMembers) {
if (!memberId.equals(senderId)) {
sessionManager.sendToUser(memberId,
JsonUtil.toJson(GroupNotification.newMsg(groupId, msgId)));
}
}
}
return msgId;
}
}
// ========== MQ消费者：处理写扩散 ==========
@RocketMQMessageListener(topic = "group-fanout", consumerGroup = "fanout-consumer")
public class FanoutMessageConsumer implements RocketMQListener<FanoutMessage> {
@Override
public void onMessage(FanoutMessage msg) {
for (String userId : msg.getTargetUserIds()) {
if (userId.equals(msg.getSenderId())) continue;
// 写入用户收件箱
String inboxKey = "inbox:" + userId;
redisTemplate.opsForZSet().add(inboxKey, msg.getMsgId(),
System.currentTimeMillis());
// 在线则实时推送
if (sessionManager.isOnline(userId)) {
deliveryService.pushMessage(msg, userId);
} else {
offlineService.storeOffline(msg, userId);
}
}
}
}
流程3：离线消息拉取代码
// 离线消息拉取 — 用户上线后执行
@Service
public class OfflineMessagePullService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private MessageRepository messageRepo;
private static final String OFFLINE_PREFIX = "offline:";
/**
* 用户上线 — 拉取离线消息
* 1. 从Redis List取出离线消息ID列表
* 2. 批量从DB查消息详情（避免Redis存完整消息体太大）
* 3. 按时间排序后返回
*/
public List<Message> pullOnLogin(String userId) {
String key = OFFLINE_PREFIX + userId;
// 取出所有离线消息ID
List<String> msgIds = redisTemplate.opsForList().range(key, 0, -1);
if (msgIds == null || msgIds.isEmpty()) {
return Collections.emptyList();
}
// 批量查消息详情
List<Message> messages = messageRepo.findByIds(msgIds);
// 按时间排序
messages.sort(Comparator.comparingLong(Message::getTimestamp));
// 清空离线队列
redisTemplate.delete(key);
// 异步：标记这些消息为已送达
CompletableFuture.runAsync(() -> {
messages.forEach(msg ->
messageRepo.updateStatus(msg.getId(), DeliveryStatus.DELIVERED));
});
return messages;
}
/**
* 增量拉取 — 客户端记录最后消息ID，只拉取新增的
* 适用于长时间在线后的增量同步
*/
public List<Message> pullIncremental(String userId, String lastMsgId) {
// 查询lastMsgId之后的所有消息
return messageRepo.findByReceiverAfterId(userId, lastMsgId);
}
}
💥 踩坑实录 — 4个真实事故
坑1：消息乱序 — 网络延迟导致消息顺序错乱
现象：用户A发了3条消息"1""2""3"，对方收到顺序变成了"1""3""2"。
原因：消息2在网络中延迟了，消息3先到达。就像3封信走不同邮路，先发的后到。
发"1"
发"2"
发"3"
↓ 快
↓ 慢(网络延迟)
↓ 快
收"1"
收"3"
收"2"
解决方案：给每条消息加序号(seq)，客户端按seq排序显示。
// 消息序号方案
public class MessageSeqService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 为会话分配消息序号
* 每个会话独立递增，保证会话内有序
*/
public long nextSeq(String sessionId) {
String key = "msg:seq:" + sessionId;
return redisTemplate.opsForValue().increment(key);
}
}
// 客户端：按seq排序显示
// messages.sort((a, b) => a.seq - b.seq);
// 即使收到顺序乱了，展示也是正确的
✅ 关键：消息序号保证逻辑顺序，不依赖网络到达顺序。就像快递单号——3号包裹可能比2号先到，但你按单号排序就不会搞错。
坑2：群消息风暴 — 500人群每秒100条消息
现象：活跃群里消息刷屏，服务端CPU飙升，推送延迟严重。
原因：500人 × 100条/秒 = 每秒5万次推送。就像500人同时说话，每个人都要听其他499人的——信息爆炸。
解决方案：限流 + 合并推送
// 群消息限流 + 合并推送
@Service
public class GroupMessageThrottleService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
// 每群每秒最大消息数
private static final int MAX_MSG_PER_SEC = 50;
// 合并推送窗口：200ms内的消息合并成一条推送
private static final long MERGE_WINDOW_MS = 200;
// 合并缓冲区: groupId → 待推送消息列表
private final ConcurrentHashMap<String, List<Message>> mergeBuffer
= new ConcurrentHashMap<>();
/**
* 限流检查
*/
public boolean allowMessage(String groupId) {
String key = "group:rate:" + groupId;
Long count = redisTemplate.opsForValue().increment(key);
if (count == 1) {
// 第一条消息，设置1秒过期
redisTemplate.expire(key, 1, TimeUnit.SECONDS);
}
return count <= MAX_MSG_PER_SEC;
}
/**
* 合并推送
* 200ms内的消息打包成一条推送，减少WebSocket帧数
*/
public void addToMergeBuffer(Message msg, String groupId) {
mergeBuffer.computeIfAbsent(groupId, k -> new CopyOnWriteArrayList<>())
.add(msg);
// 延迟200ms后批量推送
scheduler.schedule(() -> flushBuffer(groupId),
MERGE_WINDOW_MS, TimeUnit.MILLISECONDS);
}
private void flushBuffer(String groupId) {
List<Message> batch = mergeBuffer.remove(groupId);
if (batch != null && !batch.isEmpty()) {
// 合并成一条WebSocket帧发送
String json = JsonUtil.toJson(MergedMessage.of(batch));
sessionManager.broadcastToGroup(groupId, json);
}
}
}
坑3：WebSocket连接假死 — 网络断但服务端不知道
现象：客户端已经断网了，服务端的Session还在，消息推送过去以为成功，实际对方根本没收到。
原因：TCP连接的"半开"状态——一方断网，另一方不知道。就像打电话对方手机没电了，你还在说话，但对方已经听不到了。
解决方案：双向心跳检测 + 连接有效性验证
// 服务端心跳检测 — 使用Netty的IdleStateHandler
// 在Pipeline中添加：
pipeline.addLast(new IdleStateHandler(60, 30, 0, TimeUnit.SECONDS));
// 参数说明：
// readerIdleTime=60: 60秒没读到数据 → 触发空闲事件 → 关闭连接
// writerIdleTime=30: 30秒没写数据 → 触发空闲事件 → 发心跳
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
switch (event.state()) {
case READER_IDLE:
// 读空闲：客户端可能已断网，关闭连接
log.warn("读空闲超时，关闭连接: {}", ctx.channel());
ctx.close();
break;
case WRITER_IDLE:
// 写空闲：主动发心跳探测连接
ctx.writeAndFlush(new TextWebSocketFrame(
"{\"type\":\"ping\"}"));
break;
}
}
}
💡 最佳实践：客户端30秒发一次心跳，服务端60秒收不到心跳就关闭连接。客户端发现连接断了立即重连。这样即使出现"假死"连接，最多60秒就能检测到并清理。
坑4：离线消息堆积 — 用户3个月没上线
现象：某用户3个月没上线，Redis里存了大量离线消息，占用大量内存。
原因：离线消息只存不清理，越来越多。就像信箱不清理，信越堆越多。
解决方案：过期清理策略
// 离线消息过期清理策略
@Service
public class OfflineMessageCleanService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private MessageRepository messageRepo;
// 离线消息保留天数
private static final int RETENTION_DAYS = 7;
// 每个用户最大离线消息数
private static final int MAX_OFFLINE_COUNT = 500;
/**
* 定时清理过期离线消息 — 每天凌晨3点执行
*/
@Scheduled(cron = "0 0 3 * * ?")
public void cleanExpiredOfflineMessages() {
// 扫描所有离线消息Key
Set<String> keys = redisTemplate.keys("offline:*");
for (String key : keys) {
Long size = redisTemplate.opsForList().size(key);
if (size != null && size > MAX_OFFLINE_COUNT) {
// 超过500条，只保留最新的500条
long removeCount = size - MAX_OFFLINE_COUNT;
for (long i = 0; i < removeCount; i++) {
redisTemplate.opsForList().leftPop(key);
}
log.info("清理离线消息: key={}, 移除{}条", key, removeCount);
}
// 刷新过期时间
redisTemplate.expire(key, RETENTION_DAYS, TimeUnit.DAYS);
}
}
/**
* 替代方案：存入离线消息时直接限制
* 比定时清理更实时
*/
public void storeWithLimit(Message msg) {
String key = "offline:" + msg.getReceiverId();
redisTemplate.opsForList().rightPush(key, JsonUtil.toJson(msg));
// 裁剪到最新500条
redisTemplate.opsForList().trim(key, -MAX_OFFLINE_COUNT, -1);
redisTemplate.expire(key, RETENTION_DAYS, TimeUnit.DAYS);
}
}