分布式场景下如何保证消息队列实现最终一致性

Posted 2022-01-07

考虑一个分布式场景中一个常见的场景：服务A执行某个数据库操作成功后，会发送一条消息到消息队列，现在希望只有数据库操作执行成功才发送这条消息。下面是一些常见的作法：

1. 先执行数据库操作，再发送消息

public void purchaseOrder() 
    orderDao.save(order);
    messageQueue.send(message);

有可能order新增成功，发送消息失败。最终形成不一致状态。

2. 先发送消息，再执行数据库操作

public void purchaseOrder() 
    messageQueue.send(message);
    orderDao.save(order);

有可能消息发送成功，而order新增失败，从而形成不一致状态。

3. 在数据库事务中，先发送消息，再执行数据库操作

@Transactional public void purchaseOrder() 
    messageQueue.send(message);
    orderDao.save(order);

这里同样无法保证一致性。如果数据库操作成功，然而消息已经发送了，无法进行回滚。

4. 在数据库事务中，先执行数据库操作，再发送消息

@Transactional public void purchaseOrder() 
    orderDao.save(order);
    messageQueue.send(message);

这种方案成功与否，取决于消息队列是否拥有应答机制和事务机制。

应答机制表示producer发送消息后，消息队列能够返回response从而证明消息是否插入成功。

如果消息队列拥有应答机制，将上面的代码改写为：

@Transactional public void purchaseOrder() 
    orderDao.save(order); try
        kafkaProducer.send(message).get();
     catch(Exception e) throw new RuntimeException("Fail to send message");
    

这段代码表示如果发送发收到消息队列错误的response，就抛出一个RuntimeException。那么消息发送失败，能够造成数据库操作的回滚。这个方案看似可行，然而存在这样一种情况，如果消息发送成功，而消息队列由于网络原因没有即时返回response，此时消息发送方由于没有及时收到应答从而认为消息发送失败了，因此消息发送方的数据库事务回滚了，然而消息的确已经插入成功，从而造成了最终不一致性。

上面的不一致性可以通过消息的事务机制解决。

事务机制表示消息队列中的消息是否拥有状态，从而决定消费者是否消费该条消息。

Alibaba旗下的开源消息队列RocketMQ以高可用性闻名，它是最早支持事务消息的消息队列。Kafka从版本0.11开始也支持了事务机制。

RoketMQ的事务机制是将消息标记为Prepared状态或者Confirmed状态。处于Prepared状态的消息对consumer不可见。

而Kafka通过Transaction Marker将消息标记为Uncommited或Commited状态。Consumer通过配置isolation-level为read_committed或read_uncommitted来决定对哪种类型的消息可见。

5. 消息队列不支持事务消息

如果消息队列不支持事务消息，那么我们的解决方案是，新增一张message表，并开启一个定时任务扫描这张message表，将所有状态为prepared的message发送给消息队列，发送成功后，将message状态置为confirmed。

代码如下：

@Transactional public void purchaseOrder() 
    orderDao.save(order);
    messageService.save(message);

此时插入order和插入message的逻辑处于同一个数据库事务，通过后台的定时程序不断扫描message表，因此一定能够保证消息被成功投递到消息消费方。

这个方案存在的一个问题是，有可能后台任务发送消息成功后宕机了，从而没有来得及将已发送的message状态置为confirmed。因此下一次扫描message表时，会重复发送该条消息。这就是at least once delivery。

由于at least once delivery的特性，consumer有可能收到重复的数据。此时可以在consumer端建立一张message_consume表，来判断消息是否已经消费过，如果已经消费过，那么就直接丢弃该消息。

---

本文的重点是你有没有收获与成长，其余的都不重要，希望读者们能谨记这一点。同时我经过多年的收藏目前也算收集到了一套完整的学习资料，包括但不限于：分布式架构、高可扩展、高性能、高并发、Jvm性能调优、Spring，MyBatis，Nginx源码分析，Redis，ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多个知识点高级进阶干货，希望对想成为架构师的朋友有一定的参考和帮助

需要更详细架构师技能思维导图和以下资料的可以加一下技术交流分享群：“708 701 457”免费获取