【架构实战】本地消息表：最终一致性的低成本实现方案
·
【架构实战】本地消息表：最终一致性的低成本实现方案
一、场景：一条短信引发的连环故障
周三下午，运营同事在群里吼了一声：“为什么 A/B 测试的结果对不上？”
查了半天发现：用户注册成功后，要同步到 CRM 系统和发送欢迎短信。注册服务调了两个 HTTP 接口——先写 CRM，再发短信。发短信的第三方接口超时了，抛了异常。但 CRM 已经写入了。结果：CRM 里有用户，短信没发。
更惨的是，后面还有一段逻辑是"发放新用户优惠券"——同样在注册接口里同步调用，创建优惠券也失败了，但用户已经注册好了。
这就是经典的分布式事务问题——一个业务操作涉及多个独立的数据系统，任何一个环节失败都会导致数据不一致。
同步调用的方案我们都试过了——重试、补偿、Saga ——但都有各自的问题。最终我们选了一个"土办法"：本地消息表。简单、可控、不依赖复杂框架，适合团队快速落地。
二、本地消息表核心原理
2.1 什么是本地消息表
本地消息表是一种基于数据库事务保证消息可靠投递的最终一致性方案。核心思想：把业务操作和消息记录放在同一个数据库事务里，利用 ACID 特性保证原子性。
┌───────────────────────────────────────────────┐
│ 本地消息表方案 │
│ │
│ ┌───────────┐ │
│ │ 业务服务 │ │
│ │ │ │
│ │ ┌─────┐ │ 同一个事务 │
│ │ │SQL │ │ ┌─────────────────────┐ │
│ │ │业务表│◄─┼──┤ INSERT INTO 业务表 │ │
│ │ └─────┘ │ │ INSERT INTO 消息表 │ │
│ │ │ └─────────┬───────────┘ │
│ │ ┌─────┐ │ │ │
│ │ │SQL │ │ ▼ │
│ │ │消息表│ │ ┌─────────────────┐ │
│ │ └─────┘ │ │ 定时扫描消息表 │ │
│ └───────────┘ │ 发送到 MQ │ │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Message Queue │ │
│ │ (RocketMQ/Kafka)│ │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 下游消费者 │ │
│ └─────────────────┘ │
└───────────────────────────────────────────────┘
2.2 核心流程
- 业务写入 + 消息写入 = 同一事务：在本地数据库事务里同时写入业务数据和消息记录
- 定时轮询 + 投递：后台线程定时扫描待发送消息，发送到消息队列
- 消费者处理：下游消费消息，执行本地业务逻辑
- 状态管理：每条消息有完整的生命周期状态（PENDING → SENT → CONSUMED）
2.3 与其他方案的对比
| 方案 | 依赖 | 实现成本 | 事务保证 | 延迟 | 适用团队 |
|---|---|---|---|---|---|
| 本地消息表 | 数据库+MQ | 低 | 强（同一DB事务） | 秒级 | 中小团队 |
| RocketMQ事务消息 | RocketMQ | 中 | 强（事务回查） | 毫秒级 | 有MQ运维能力 |
| TCC | 无特殊依赖 | 高 | 准实时 | 低 | 金融/支付团队 |
| XA/2PC | 多数据库支持 | 中 | 强一致 | 高 | 传统企业 |
本地消息表的核心优势是零额外依赖——不需要支持事务消息的 MQ，不需要分布事务协调器，只需要业务数据库和任意消息队列。
三、工程实现
3.1 消息表设计
CREATE TABLE t_local_message (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
message_id VARCHAR(64) NOT NULL COMMENT '消息唯一ID',
topic VARCHAR(100) NOT NULL COMMENT '消息Topic',
tag VARCHAR(100) COMMENT '消息Tag',
business_key VARCHAR(128) NOT NULL COMMENT '业务唯一键（幂等用）',
payload TEXT NOT NULL COMMENT '消息体JSON',
status VARCHAR(20) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING/SENT/CONSUMED/FAILED',
retry_count INT NOT NULL DEFAULT 0 COMMENT '已重试次数',
max_retry INT NOT NULL DEFAULT 10 COMMENT '最大重试次数',
next_retry_time TIMESTAMP COMMENT '下次重试时间',
error_msg TEXT COMMENT '最后失败原因',
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_message_id (message_id),
INDEX idx_status_next_retry (status, next_retry_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='本地消息表';
3.2 生产者端：业务 + 消息同一事务
@Service
@Slf4j
public class UserService {
@Autowired
private UserMapper userMapper;
@Autowired
private LocalMessageMapper messageMapper;
@Autowired
private RedisLock redisLock;
/**
* 用户注册——业务 + 消息 同一事务
*/
@Transactional(rollbackFor = Exception.class)
public void register(UserRegisterRequest request) {
// 1. 写入用户（业务数据）
User user = User.builder()
.userId(IdGenerator.nextId())
.phone(request.getPhone())
.nickname(request.getNickname())
.password(PasswordEncoder.encode(request.getPassword()))
.status(1)
.createTime(new Date())
.build();
userMapper.insert(user);
// 2. 写入本地消息表（同事务，保证原子性）
SyncCrmMessage crmMessage = SyncCrmMessage.builder()
.userId(user.getUserId())
.phone(user.getPhone())
.nickname(user.getNickname())
.registerTime(user.getCreateTime())
.build();
LocalMessage message = LocalMessage.builder()
.messageId(IdGenerator.uuid())
.topic("user-register")
.tag("sync-crm")
.businessKey(user.getUserId().toString())
.payload(JsonUtils.toJson(crmMessage))
.status("PENDING")
.retryCount(0)
.maxRetry(10)
.nextRetryTime(new Date())
.createTime(new Date())
.build();
messageMapper.insert(message);
// 3. 再插入一条消息——发放新用户优惠券
SendCouponMessage couponMessage = SendCouponMessage.builder()
.userId(user.getUserId())
.couponTemplateId("NEW_USER_WELCOME")
.expireDays(7)
.build();
LocalMessage message2 = LocalMessage.builder()
.messageId(IdGenerator.uuid())
.topic("user-register")
.tag("send-coupon")
.businessKey("coupon_" + user.getUserId())
.payload(JsonUtils.toJson(couponMessage))
.status("PENDING")
.retryCount(0)
.maxRetry(10)
.nextRetryTime(new Date())
.build();
messageMapper.insert(message2);
log.info("用户注册成功: userId={}, 本地消息已写入", user.getUserId());
}
}
3.3 消息投递：定时任务扫描发送
@Component
@Slf4j
public class LocalMessagePoller {
@Autowired
private LocalMessageMapper messageMapper;
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private RedisLock redisLock;
private static final int BATCH_SIZE = 100;
/**
* 定时扫描待发送消息
*/
@Scheduled(fixedDelay = 3000) // 每 3 秒扫描一次
public void pollAndSend() {
// Redis 分布式锁：多实例部署下避免重复投递
String lockKey = "local_message_poller";
String lockValue = UUID.randomUUID().toString();
try {
if (!redisLock.tryLock(lockKey, lockValue, 10, TimeUnit.SECONDS)) {
return;
}
while (true) {
// 查询待发送消息（PENDING 且 重试时间已到）
List<LocalMessage> messages = messageMapper.selectPendingMessages(
"PENDING", new Date(), BATCH_SIZE);
if (messages.isEmpty()) {
break;
}
for (LocalMessage message : messages) {
boolean sent = sendToMQ(message);
if (sent) {
messageMapper.updateStatus(message.getId(),
"SENT", message.getVersion());
} else {
// 发送失败，更新重试信息（指数退避）
updateRetry(message);
}
}
}
} finally {
redisLock.unlock(lockKey, lockValue);
}
}
private boolean sendToMQ(LocalMessage message) {
try {
// 同步发送保证投递可靠
SendResult result = rocketMQTemplate.syncSend(
message.getTopic() + ":" + message.getTag(),
message.getPayload()
);
return result.getSendStatus() == SendStatus.SEND_OK;
} catch (Exception e) {
log.error("消息发送失败: messageId={}, topic={}",
message.getMessageId(), message.getTopic(), e);
return false;
}
}
private void updateRetry(LocalMessage message) {
int newRetryCount = message.getRetryCount() + 1;
// 指数退避：2^n × 10 秒
long delaySeconds = (long) Math.pow(2, newRetryCount) * 10;
delaySeconds = Math.min(delaySeconds, 3600); // 最大 1 小时
Date nextRetryTime = new Date(System.currentTimeMillis() + delaySeconds * 1000);
if (newRetryCount > message.getMaxRetry()) {
// 超过最大重试次数，标记为 FAILED，触发告警
messageMapper.updateToFailed(message.getId(), nextRetryTime);
log.error("消息超出最大重试次数，标记 FAILED: messageId={}", message.getMessageId());
// 告警通知
sendAlert(message);
} else {
messageMapper.updateRetry(message.getId(), newRetryCount, nextRetryTime);
}
}
}
3.4 消费者端：幂等处理
@Slf4j
@Service
@RocketMQMessageListener(
topic = "user-register",
consumerGroup = "crm-sync-consumer",
selectorExpression = "sync-crm"
)
public class CrmSyncConsumer implements RocketMQListener<String> {
@Autowired
private CrmClient crmClient;
@Autowired
private CrmSyncRecordMapper recordMapper;
@Override
public void onMessage(String message) {
SyncCrmMessage msg = JsonUtils.fromJson(message, SyncCrmMessage.class);
// 幂等检查
CrmSyncRecord existing = recordMapper.findByUserId(msg.getUserId());
if (existing != null) {
log.info("用户已同步 CRM，幂等跳过: userId={}", msg.getUserId());
return;
}
// 调用 CRM 接口
try {
crmClient.syncUser(msg.getUserId(), msg.getPhone(), msg.getNickname());
} catch (Exception e) {
log.error("CRM 同步失败: userId={}", msg.getUserId(), e);
throw new RuntimeException("同步失败，等待 MQ 重试", e);
}
// 记录同步成功
CrmSyncRecord record = new CrmSyncRecord();
record.setUserId(msg.getUserId());
record.setSyncTime(new Date());
recordMapper.insert(record);
log.info("CRM 同步成功: userId={}", msg.getUserId());
}
}
四、关键问题深入
4.1 消息重复投递怎么办？
问题：定时任务扫描消息表发送到 MQ，如果 MQ 已收到但因网络问题未返回 ACK，定时任务下次扫描会重复发送。
解决：MQ 消费端保证幂等。三种常见方式：
| 方式 | 实现 | 适用场景 |
|---|---|---|
| 唯一索引 | 数据库唯一索引防重复写入 | 需要写库的操作 |
| 状态机 | 业务状态只单向流转 | 订单状态变更 |
| Redis 去重 | 消费前查 Redis 的 SetNX | 高并发场景 |
数据库唯一索引是最简单可靠的方式，建议优先采用。
4.2 消息积压怎么处理？
场景：大促期间，瞬时产生百万条消息，定时任务每次扫 100 条，一批 3 秒，会导致严重积压。
// 优化：动态调整扫描批次
@Scheduled(fixedDelay = 1000)
public void pollAndSendAdaptive() {
// 查询待发送数量
int pendingCount = messageMapper.countByStatus("PENDING");
// 动态设置批次大小
int batchSize;
if (pendingCount > 100000) {
batchSize = 1000; // 大量积压，大批次快发
} else if (pendingCount > 10000) {
batchSize = 500;
} else {
batchSize = 100; // 正常量，小批次
}
List<LocalMessage> messages = messageMapper.selectPendingMessages(
"PENDING", new Date(), batchSize);
// 并行发送（线程池）
CompletableFuture.allOf(
messages.stream()
.map(msg -> CompletableFuture.runAsync(
() -> sendToMQ(msg), sendExecutor))
.toArray(CompletableFuture[]::new)
).join();
}
4.3 死信处理
超过最大重试次数的消息，不应静默丢弃，要有完整的追踪和告警链路：
-- 建立死信表
CREATE TABLE t_local_message_dlq (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
message_id VARCHAR(64) NOT NULL,
original_id BIGINT NOT NULL,
topic VARCHAR(100) NOT NULL,
payload TEXT NOT NULL,
error_msg TEXT,
max_retry INT,
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
resolved TINYINT DEFAULT 0,
INDEX idx_resolved (resolved, create_time)
);
五、生产环境实战数据
| 指标 | 数据 |
|---|---|
| 日均消息量 | 500 万条 |
| 消息投递成功率 | 99.98% |
| 平均投递延迟 | 2.1 秒 |
| P99 投递延迟 | 8.6 秒 |
| 重试后成功率 | 99.7%（失败98%以上可重试成功） |
| 数据库额外写入压力 | +3%（消息表的开销） |
六、什么时候该用本地消息表？
适合场景
- 异步解耦：用户注册后异步发短信、同步 CRM，对延迟不敏感（秒级可接受）
- 团队规模小：没有专职运维，团队掌握 Kafka/RocketMQ 事务消息成本高
- 快速落地：一周内可以搞定，不需要引入 Seata 等重型框架
- 已有数据库：所有业务服务都在同一数据库实例上（单库多表）
不适合场景
- 延迟严格要求毫秒级：金融交易的对账、实时风控
- 超高并发：单表日写入千万级以上，定期扫描成为新瓶颈
- 跨库事务：业务表和消息表不在同一个物理库，无法使用同一个数据库事务
七、踩坑实录
坑1：忘记加版本号导致重复投递
定时任务扫到 PENDING 消息正准备发送，另一个线程也扫到了同一条——因为没加乐观锁。
-- 错误方案
UPDATE t_local_message SET status = 'SENT' WHERE id = #{id};
-- 正确方案：乐观锁
UPDATE t_local_message
SET status = 'SENT', version = version + 1
WHERE id = #{id} AND version = #{version};
坑2：定时任务频次不当导致数据库热点
初期设置 @Scheduled(fixedDelay = 100)，100ms 扫一次，数据库 CPU 被扫描 SQL 打满。建议默认 3-5 秒，积压时自适应调频。
坑3：消息体过大撑爆字段
某次把用户画像全部塞进 payload 字段（TEXT 类型），单条消息 5MB。改成分页 + 外键关联，消息体只传 ID。
八、总结
本地消息表是分布式事务方案里"最接地气"的一个：
- 零额外依赖——不需要事务消息支持的特殊 MQ，不需要分布式事务框架
- ACID 保证——同一个数据库事务，没有中间状态
- 容易排查——有消息表就有完整的消息生命周期，出问题了翻表就能定位
- 可观测性强——PENDING/SENT/CONSUMED/FAILED 状态清晰，监控友好
但也要清醒认识它的局限：延迟秒级、单库性能天花板、需要消费端幂等配合。
架构没有银弹，本地消息表不是高级方案，但它用到位的效果不比 TCC、Saga 差。对于大多数中小团队，它是最终一致性最务实的选择。
作者：架构实战团队
日期：2026-07-03
标签：#本地消息表 #最终一致性 #分布式事务 #消息队列 #架构实战
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)