zoukankan      html  css  js  c++  java
  • 分布式场景下如何保证消息队列实现最终一致性

    考虑一个分布式场景中一个常见的场景:服务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-levelread_committedread_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表,来判断消息是否已经消费过,如果已经消费过,那么就直接丢弃该消息。

  • 相关阅读:
    Windows Phone 播放音频之MediaPlayer
    (转)系统架构师学习笔记_第四章(下)
    (转)系统架构师学习笔记_第二章
    (转)系统架构师学习笔记_第四章(上)
    读取properties配置文件信息
    (转)系统架构师学习笔记_第五章(上)
    (转)系统架构师学习笔记_第三章
    Struts2实现文件上传
    Struts2手工编写代码实现对Action中所有方法输入校验
    (转)系统架构师学习笔记_第五章(下)
  • 原文地址:https://www.cnblogs.com/lfs2640666960/p/11012816.html
Copyright © 2011-2022 走看看