当前位置：首页 > news >正文

MQ的数据一致性，如何保证？

news 来源：原创 2025/5/2 7:37:30

1 数据一致性问题的原因

这些年在Kafka、RabbitMQ、RocketMQ踩过的坑，总结成四类致命原因：

生产者悲剧：消息成功进Broker，却没写入磁盘就断电。
消费者悲剧：消息消费成功，但业务执行失败。
轮盘赌局：网络抖动导致消息重复投递。
数据孤岛：数据库和消息状态割裂

这些情况，都会导致MQ产生数据不一致的问题。

2 消息不丢的方案

我们首先需要解决消息丢失的问题。

2.1 事务消息的两阶段提交
以RocketMQ的事务消息为例，伪代码如下：

// 发送事务消息核心代码
TransactionMQProducer producer = new TransactionMQProducer("group");
producer.setTransactionListener(new TransactionListener() {
    // 执行本地事务（比如扣库存）
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        return doBiz() ? LocalTransactionState.COMMIT : LocalTransactionState.ROLLBACK;
    }

    // Broker回调检查本地事务状态
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return checkDB(msg.getTransactionId()) ? COMMIT : ROLLBACK;
    }
});

真实场景中，别忘了在checkLocalTransaction里做好妥协查询（查流水表或分布式事务日志）。

2.2 持久化配置
RabbitMQ的坑都在配置表里：

在这里插入图片描述

RabbitMQ本地存储+备份交换机双重保护代码如下：

channel.queueDeclare("order_queue", true, false, false, 
    new HashMap<String, Object>(){{
        put("x-dead-letter-exchange", "dlx_exchange"); // 死信交换机
    }});

2.3 副本配置
在这里插入图片描述

server.properties配置如下：

acks=all
min.insync.replicas=2
unclean.leader.election.enable=false

3 应对重复消费的方案

接下来，需要解决消息的重复消费问题。

3.1 唯一ID

// 雪花算法生成全局唯一ID
Snowflake snowflake = new Snowflake(datacenterId, machineId);
String bizId = "ORDER_" + snowflake.nextId();

// 查重逻辑（Redis原子操作）
String key = "msg:" + bizId;
if(redis.setnx(key, "1")) {
    redis.expire(key, 72 * 3600);
    processMsg();
}

先使用雪花算法生成全局唯一ID，然后使用Redis的setnx命令加分布式锁，来保证请求的唯一性。

本地布隆过滤器+分布式Redis 双校验

3.2 幂等设计
针对不同业务场景的三种对策：
在这里插入图片描述

public void addPoints(String userId, String orderId, Long points) {
    if (pointLogDao.exists(orderId)) return;
    
    User user = userDao.selectForUpdate(userId); // 悲观锁
    user.setPoints(user.getPoints() + points);
    userDao.update(user);
    pointLogDao.insert(new PointLog(orderId)); // 幂等日志
}

这里使用了数据库行锁实现的幂等性。

3.3 死信队列
RabbitMQ的配置如下：

// 消费者设置手动ACK
channel.basicConsume(queue, false, deliverCallback, cancelCallback);

// 达到重试上限后进入死信队列
public void process(Message msg) {
    try {
        doBiz();
        channel.basicAck(deliveryTag);
    } catch(Exception e) {
        if(retryCount < 3) {
            channel.basicNack(deliveryTag, false, true);
        } else {
            channel.basicNack(deliveryTag, false, false); // 进入DLX
        }
    }
}

消费者端手动ACK消息。

在消费者端消费消息时，如果消费失败次数，达到重试上限后进入死信队列。

可以通过DLX收集全部异常消息，凌晨用补偿Job重跑。

4 系统架构设计

接下来，从系统架构设计的角度，聊聊MQ要如何保证数据一致性？

4.1 生产者端
对于实效性要求不太高的业务场景，可以使用：本地事务表+定时任务扫描的补偿方案。

流程图如下：
在这里插入图片描述

4.2 消费者端
消费者端为了防止消息风暴，要设置合理的并发消费线程数。

流程图如下：
在这里插入图片描述

4.3 终极方案
对于实时性要求比较高的业务场景，可以使用事务消息+本地事件表的黄金组合.

流程图如下：
在这里插入图片描述

5 经验十条
1、消息必加唯一业务ID（别用MQ自带的ID）
2、消费逻辑一定要幂等（重复消费是必然事件）
3、数据库事务和消息发送必须二选一（或者用事务消息）
4、消费者线程数不要超过分区数*2
5、死信队列必须加监控报警
6、测试环境一定要模拟网络抖动（chaos engineering）
7、消息体要兼容版本号
8、不要用消息队列做业务主流程
9、消费者offset定时存库（防止重平衡丢消息）
10、业务指标和MQ监控要联动（比如订单量和消息量的波动要同步）