消息可靠性：用 outbox 模式把 IM 消息投递变得靠谱
做即时通讯的时候，最怕的不是消息发不出去，而是消息发出去了对方没收到——你以为通知到位了，对方以为你没通知。
我们在 Workforce Hub 的 IM 模块里踩了不少坑，最后用 outbox 模式 + 定时调度 + 死信队列的思路把可靠性搭了出来。这篇就把这条路走一遍。
一、问题在哪？——WebSocket 推送的三个坑
上一篇讲了用 WebSocket 做实时推送，看起来挺简单的：服务端 session.sendMessage() 一发，客户端就收到了。
但实际跑起来，问题一个接一个。下面是我踩过的三个最深坑——
坑一：双写不一致——事务还没提交，推送已经发出去了
第一版代码大概是这样的：
// 第一版：天真模式
@Transactional
public void sendMessage(SendImMessageRequest req) {
ImMessage saved = messageRepository.insert(req); // ① INSERT
wsSession.sendToUser(req.toUserId(), saved); // ② 推送
// ③ 如果后面抛异常，事务回滚 →
// 但是②已经推了！用户看到了，刷新页面又没了
}
问题本质：insert 走 JDBC 事务，sendToUser 走 WebSocket，两者不在同一个原子边界里。
具体踩坑场景：
@Transactional
public void sendMessage(SendImMessageRequest req) {
ImMessage saved = messageRepository.insert(req);
wsSession.sendToUser(req.toUserId(), saved);
// 这里抛异常了！比如敏感词检测不过
contentModerationService.check(saved.getContent());
// → 事务回滚，saved 不存在了
// → 但 wsSession.sendToUser 已经发出去了！
}
用户 A 看到消息发出去了（绿色 ），用户 B 刷新后消息消失——“你是不是把消息撤回了？”“没有啊我发了的。”
不是 bug，是架构缺陷——两个写操作（数据库 + WebSocket）天然无法在一个事务里。
坑二：离线就丢——静默失败，没人知道
// 第二版：加了状态判断，但没解决根本问题
public void pushToUsers(List<String> userIds, ImWebSocketEvent event) {
for (String userId : userIds) {
String sessionId = sessionRegistry.getSessionId(userId);
if (sessionId == null) {
log.warn("用户 {} 不在线，跳过推送", userId);
return; // ← 直接 return！消息丢了！
}
wsSession.sendToUser(sessionId, event);
}
}
用户 A 发给用户 B，B 手机没电关机了。sessionRegistry 里没有 B 的会话 → log.warn 打了一行日志 → 没了。
没有重试、没有离线队列、没有任何兜底。用户 B 开机后永远不知道 A 说了什么。
这个坑比坑一更严重——坑一至少能手动重发，坑二是直接丢了，连重发的机会都没有。
坑三：服务重启——内存里的"没推完"全丢了
// 推送状态只能在内存里，靠 Map<sessionId, wsSession> 持有
// 一旦进程重启：
// - WebSocket 连接全断
// - 正在推送中的消息丢失
// - 还没来得及推送的也丢了
// - MySQL 里数据还在，但没人知道"这些消息还没推"
画个时间线：
12:00:00 用户 A 发消息 → INSERT 成功
12:00:01 还没来得及 push → OOM Kill → JVM 进程挂
12:00:05 新进程启动 → 没有任何记录表明"这条消息没推"
12:30:00 用户 B 上线 → 看不到 A 的消息
这条消息变成了「僵尸消息」——数据库里有，但没有任何机制能发现它还没送达。
三个坑的共同解
三个坑串起来看，本质是同一个问题：推送操作不可靠，但没有持久化的"投递状态"来追踪。
一句话总结：推送不应该发生在主业务流程里，它应该是一个独立的、有状态追踪的异步过程。
二、outbox 模式——先把信放进邮筒
outbox 模式的核心思想很简单：不直接推，先写表。
整个链路变成三步：
写入：把推送事件写入
im_event_outbox表，跟消息数据在同一个事务里调度：独立的定时任务从 outbox 表拉取待投递事件
投递：逐个推送，成功标 SENT，失败标 FAILED + 安排重试
outbox 表结构（DDL + 索引踩坑）
CREATE TABLE im_event_outbox (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
event_type VARCHAR(32) NOT NULL COMMENT 'message.created / conversation.updated',
conversation_id BIGINT NOT NULL,
message_id BIGINT NOT NULL COMMENT '关联消息ID',
seq_no BIGINT NOT NULL COMMENT '消息序号，保证顺序',
sender_id BIGINT NOT NULL,
receiver_id BIGINT NOT NULL COMMENT '接收方用户ID',
payload JSON NOT NULL COMMENT '消息体快照，推送时直接用',
status VARCHAR(16) NOT NULL DEFAULT 'PENDING'
COMMENT 'PENDING → SENDING → SENT/FAILED/DEAD',
retry_count INT NOT NULL DEFAULT 0,
next_retry_at DATETIME NOT NULL COMMENT '下次可重试时间',
claimed_at DATETIME NULL COMMENT '被调度器认领的时间',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP,
-- ⭐ 最关键索引：调度器每2秒扫一次
INDEX idx_status_retry (status, next_retry_at),
-- 辅助索引：按接收方查投递历史
INDEX idx_receiver_time (receiver_id, created_at),
-- 辅助索引：按消息ID查投递状态
INDEX idx_message_id (message_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
索引选择的血泪教训：
idx_status_retry(status, next_retry_at) 是最重要的索引——调度器的核心查询是：
SELECT id FROM im_event_outbox
WHERE (status IN ('PENDING', 'FAILED') AND next_retry_at <= NOW())
OR (status = 'SENDING' AND claimed_at < NOW() - INTERVAL 60 SECOND)
ORDER BY next_retry_at ASC
LIMIT 100
FOR UPDATE SKIP LOCKED
我一开始只给 status 加了单列索引，结果 next_retry_at 的范围扫描不走索引，每次全表扫描十几万行，调度周期 CPU 直接飙到 60%。加上联合索引后降到 2ms。
payload 字段为什么要 JSON 快照？ 因为 outbox 生命周期可能比消息短（死信会被清理），而且推送时不需要再 JOIN messages 表——payload 里就是完整内容，少一次查询。
写入代码
@Transactional
public void sendMessage(SendImMessageRequest req) {
// ① 存消息
ImMessage saved = messageRepository.insert(req);
// ② 构造 outbox 事件
ImWebSocketEvent event = ImWebSocketEvent.of(
"message.created", saved.conversationId(),
saved.id(), saved.seqNo(), saved.senderId(), saved
);
// ③ 写入 outbox——跟 ① 在同一个事务里
outboxService.enqueue(List.of(req.toUserId()), event);
// ④ 事务提交后才触发推送（afterCommit 回调）
pushService.pushToUsersAfterCommit(List.of(req.toUserId()), event);
}
pushToUsersAfterCommit 的实现：
public void pushToUsersAfterCommit(List<String> userIds, ImWebSocketEvent event) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCommit() {
// 事务真的提交了，才推
dispatchService.dispatch(userIds, event);
}
}
);
}
关键点：afterCommit 保证消息落库 + outbox 落库都成功了才开始推送。事务回滚 → outbox 也回滚 → 调度器扫不到 → 不会推送。坑一解决。
三、投递调度的细节——几个写代码时容易忽略的问题
光是"定时扫表 + 推送"听着简单，但细节决定可靠性。
3.1 竞争条件——多个调度器实例不能抢同一个事件
生产环境通常会跑多个实例（比如两个 pod），如果两个实例的定时任务同时抢到同一条记录，消息就会被投递两次。
用 FOR UPDATE SKIP LOCKED 解决：
SELECT id FROM im_event_outbox
WHERE (status IN ('PENDING', 'FAILED') AND next_retry_at <= ?)
OR (status = 'SENDING' AND claimed_at < ?)
ORDER BY next_retry_at ASC, id ASC
LIMIT ?
FOR UPDATE SKIP LOCKED
FOR UPDATE SKIP LOCKED 的意思是：把符合条件的行锁住，但已经被其他事务锁住的行直接跳过不阻塞。这样两个实例各抢各的，互不影响。
3.2 状态机——PENDING → SENDING → SENT/FAILED/DEAD
四种状态：PENDING（待投递）、SENDING（正在投递）、SENT（已投递）、FAILED（失败等重试）、DEAD（彻底失败）。
关于 SENDING 状态：调度器认领事件后会先 markSending，记录 claimed_at 时间。如果投递过程中调度器挂了（进程 crash），这个事件就卡在 SENDING 再也不动了。所以在抢事件的 SQL 里加了一个"防卡死"逻辑：
-- stale 检测：超过 60 秒还卡在 SENDING 的，允许其他实例捡起来
OR (status = 'SENDING' AND claimed_at < ?)
这是调度系统中常见的"认领过期"机制——超时了就允许重新争夺。
3.3 重试策略——指数退避 + 最大次数
// 每次失败延迟翻倍，最多重试 5 次，之后进死信
retryDelayMs: 5000, 20000, 80000, 320000, 1280000
maxAttempts: 5 // 超过后标记 DEAD
不需要指数退避？这里的 retryDelayMs 实际上是固定的（5 秒），因为 ImEventOutboxRepository.markFailed 直接把 nextRetryAt 设为了 now + retryDelayMs，没有翻倍逻辑——这是当前实现的一个简化点，后续可以优化。
为什么不是指数退避？ 说实话这里偷了懒——后续版本可以改成 retryDelayMs * 2^(retryCount-1)。但目前看 5 次重试 + 5 秒间隔在实际使用中已经足够覆盖瞬时网络抖动。
3.4 死信处理——重试 5 次还失败怎么办？
boolean dead = nextRetryCount >= maxAttempts;
outboxRepository.markFailed(id, dead ? "DEAD" : "FAILED", ...);
标记为 DEAD 后，事件不会自动重试，需要人工介入。后台管理页面可以查看死信列表，手动重试或丢弃。
小发现：开发阶段我把 maxAttempts 设成了 1，然后发现所有推送都进死信了——因为 WebSocket 服务器没启动。后来才意识到这个配置的含义不是"最多尝试一次"，而是"最多重试一次 = 总共两次"。所以配置项叫 maxAttempts 而不是 maxRetries。
四、整体流程——从发消息到确认送达
flowchart TD
USER[用户发送消息] --> API[POST /api/im/messages]
API --> TX[事务开始]
TX --> INSERT[ INSERT im_message]
INSERT --> OUTBOX[ INSERT im_event_outbox<br/>status=PENDING, next_retry_at=NOW]
OUTBOX --> TXCOMMIT[ 事务提交]
TXCOMMIT --> CALLBACK[ afterCommit 回调]
CALLBACK --> DISPATCH[ 立即触发投递]
TX --> ROLLBACK[ 事务回滚]
ROLLBACK --> NOPUSH[ outbox 同步回滚<br/>调度器扫不到]
DISPATCH --> WS[ WebSocket push]
WS --> SENT[ 标记 SENT]
DISPATCH --> FAIL[ 推送失败]
FAIL --> FAILED[ 标记 FAILED<br/>nextRetryAt = now + 5s]
FAILED --> RESCAN[ 下次扫描重试]
RESCAN --> DISPATCH
FAILED --> DEAD[ retryCount >= 5<br/>标记 DEAD]
DEAD --> ADMIN[ 管理员介入<br/>手动重试/丢弃]
DISPATCH -...-> OFFLINE[ 用户离线]
OFFLINE -...-> ONLINE[ 用户上线]
ONLINE -...-> RESCAN[ 下次扫描命中]
每一步都有兜底——事务回滚不推、离线下次推、失败自动重试、重试到底还有死信。
调度器的并发控制（多实例不重复抢事件）
@Scheduled(fixedDelay = 3000) // 每3秒执行
public void dispatch() {
// ① 拉取待投递事件（FOR UPDATE SKIP LOCKED）
List<ImEventOutbox> events = outboxRepository.findPending(
batchSize, sendingStaleMs
);
for (ImEventOutbox event : events) {
// ② 先标记为 SENDING，防止被其他实例抢
boolean locked = outboxRepository.markSending(
event.getId(), LocalDateTime.now()
);
if (!locked) continue; // 被其他实例抢了，跳过
try {
// ③ 推送
wsSession.sendToUser(event.getReceiverId(), event.getPayload());
// ④ 标记成功
outboxRepository.markSent(event.getId());
} catch (Exception e) {
// ⑤ 失败处理
handleFailure(event, e);
}
}
}
MySQL 的 FOR UPDATE SKIP LOCKED 是这里的关键——两个 pod 同时调度，各抢各的记录，互不阻塞。比分布式锁轻量得多，不需要引入 Redis。
性能实测
内网 4核8G 服务器、调度器单实例的实测数据：
P99 到 3.2 秒的原因：最坏情况消息到达时刚好错过一轮调度 → 等 3 秒 + 推送处理 ~200ms ≈ 3.2 秒。调小 fixedDelay 到 1 秒可以把 P99 降到 ~1.5 秒，但数据库轮询频率翻倍，需要按实际负载权衡。
五、配置一览——这些参数怎么调
workforce:
im:
outbox:
dispatcher:
enabled: true # 是否启用 outbox 调度
batch-size: 100 # 每次拉取数量
max-attempts: 5 # 最大重试次数
retry-delay-ms: 5000 # 重试间隔（毫秒）
sending-stale-ms: 60000 # SENDING 状态过期时限
fixed-delay-ms: 3000 # 调度间隔（毫秒）
调参建议：
retry-delay-ms：设太小会频繁重试压数据库，设太大用户体验差。5 秒是折中sending-stale-ms：必须大于单次投递的最大耗时，否则正常投递中的事件会被其他实例抢走。默认 60 秒足够batch-size：太小吞吐差，太大可能撑爆内存。100 条对中小企业场景够用
六、为什么不全走 MQ？——一个还没做的决定
用 RabbitMQ / Kafka 做消息投递看起来更标准：
flowchart LR
SVC[业务代码] --> MQ[(消息队列)]
MQ --> CONSUMER[消费者]
CONSUMER --> WS[WebSocket 推送]
但我们的项目目前没引入 MQ，原因是：
Docker Compose 多一个服务：小项目用 outbox + 数据库调度够用，加 MQ 反而多了运维成本
outbox 表天然可追溯：谁发的、发给谁、投了几次、失败原因，全在表里——出问题直接 SQL 查，比 MQ 的 UI 直观
后续扩展预留好了：outbox 的事件结构跟 MQ 消息体完全兼容，将来引入 Kafka 只需要把
ImWebSocketOutboxService.enqueue换成KafkaTemplate.send，调度器换成消费者即可
说实话，如果后期消息量真的大了（日活 > 1000），把 outbox 换成 Kafka 的改造成本很低，因为架构边界已经划清楚了。
默认评论
Halo系统提供的评论