• RocketMQ、Kafka、RabbitMQ 消费原理,顺序消费问题【图文理解】


    B站视频地址




    一、开始


    先来定义一下何为顺序消息,比如有A、B两条消息,消息处理的流程是 1、2、3 … 10,只有当A消息执行10完毕后,B消息才可以进行1流程。

    注:如果A执行到7,B开始执行1,这其实不一定是顺序消息,因为各种原因最终可能导致B先执行完10。


    目前比较流行的队列:RocketMQ、RabbitMQ、Kafka

    • RocketMQ 消息发送到 topic,再到topic关联的 queue
    • RabbitMQ 消息发送到 exchange,再由exchange通过规则到 queue
    • Kafka 消息发送到 topic,再到topic关联的 partitions (partitions可以理解就是一个queue)

    基于消息队列的规则,想要达到我们的目标就要求A、B两个消息先后发送到同一个 queue/partitions,且只能有一个消费者,且消费的时候必须是单线程非异步的才可满足。


    二、结果


    三种MQ都支持消息发送到指定的 queue/partition,简单来说就是基于一个标识去计算看它应该落在哪个queue/partition,同一批顺序消息的标识是一样的,所以最终进入的queue/partition也是一致的。 进入 queue/partition 之后的消息都是顺序的,它们是 FIFO的。

    顺序消息的控制主要是在消费端,那问题就变成了2个

    1. queue/partition 和消费者之间是如何对应的
    2. 消费者对同一个 queue/partition 的消息,是多线程还是单线程

    只有满足一个queue/partition 只能对应一个消费者,一个消费者对于一个queue/partition 是单线程消费的,才可以做到消费顺序。


    注:MQ有集群消费和广播消费,顺序消费肯定是建立在集群消费模式下的。

    最终结果:RocketMQ和Kafka是支持顺序消息的,RabbitMQ不支持顺序消息。


    1、RocketMQ 消费关系图


    1-1、queue和consumer的关系


    在这里插入图片描述


    从上面的图可以看到,一个queue最多只能对一个 consumer,如果某个 topic需要更大的并发,那就需要,那就增加 queue,然后增加 consumer


    1-2、consumer 和线程的关系


    正常使用SpringBoot开发项目的时候,都是引入 rocketmq-spring-boot-starter,然后用 @RocketMQMessageListener 来做消费处理,所以下面图也是基于这个用法来画的

    在这里插入图片描述


    通过这个图可以看到使用 @RocketMQMessageListener 做消费者的时候,本质上消息是被多线程去消费了,那就存在A、B消息的真正处理顺序不一致了。


    RocketMQ的解决办法是,当你设置消费为顺序消费的时候,在消息处理的时候它会基于 queue加锁,这样就只能单线程处理这个queue的消息了。


    设置顺序消费的代码

    @RocketMQMessageListener(
            topic = "Topic1", 
            consumerGroup = "springboot3_producer_group", 
            consumeMode = ConsumeMode.ORDERLY
    )
    
    • 1
    • 2
    • 3
    • 4
    • 5

    2、Kafka 消费关系图


    Kafka 里面没有queue的概念,转而用partitions,但其本质上queue和partitions是一样的,就把它理解成一个queue完事


    1-1、partitions和consumer的关系


    partitions 和 consumer的关系和 RocketMQ的一模一样,只是把queue改为partitions即可,就不画了


    1-2、consumer 和线程的关系


    在这里插入图片描述


    1. Kafa消费消息的时候是主动去拉,拉到了就去消费,消费完了,再去拉。 拉和消费的线程是一个
    2. 当自定义线程数大于 partitions 的时候,没用,这个没用的意思是 Kafka压根不会创建比分配给自己 partitions 数量更多的线程
    3. 添加消费者的时候,会自平衡(这点所有的MQ都一样的)
    4. 默认如果没有给consumer设置线程数的话,是单线程

    Kafka的解决办法是,每一个 partitions 最多只有一个线程来消费它,单线程自然就是顺序消费的咯。


    3、RabbitMQ 消费关系图


    1-1、queue和consumer的关系


    在这里插入图片描述


    1. RabbitMQ新增了exchange(交换机)的概念,所有的数据都是先发送到交换机,再由exchange基于规则下发到具体的queue
    2. 可以通过设置交换机的类型的,让消息投递到一个或多个 queue
    3. 广播消息:可以设置exchange类型为fanout,这样消息就会投递到所有与之绑定的queue(前提是没有设置特殊的 routingkey)
    4. 集群消费:可以设置多个 consumer去消费一个queue,或一个消费者设置多线程去消费,以此来增加消费速率

    注:RabbitMQ的queue和consumer是可以设置为多对多的关系


    1-2、consumer 和线程的关系


    在这里插入图片描述


    1. RabbitMQ默认也是一个线程消费
    2. 当开启了多个线程的时候,消息最终顺序就可能不一致,因为各个线程之间其实是相互独立的

    4、总结


    从上述结果来看其实三种队列都是支持顺序消息的(前提消息都发送到一个 queue/partitions),但支持的程度和结果不同

    1. RocketMQ,一个queue只能有一个consumer,消费者是多线程的,但开启顺序消费的时候,会对 queue加锁从而保证顺序
    2. Kafka,一个 partitions只能由一个consumer的一个线程去消费,基于单线程就保证了顺序性
    3. RabbitMQ,queue和consumer是多对多的,consumer的多个线程是独立的,要想保证顺序,只能让一个queue只有一个consumer,且consumer只有一个线程(但这样做效率就很低)

    三、实践


    1、全局有序


    基于上述分析,三种MQ都可以做到全局有序,因为一旦要求全局有序,消费者就必须是单线程消费。


    2、局部有序


    比如用户订单业务,对于不同的用户它们的消费顺序可以不用关注,但是对于同一个用户的消息必须是严格有序的(简单的如先下单、再支付)。

    对于这种场景RabbitMQ基本上就不满足的,它只有一个队列,如果消费者是单线程的会阻塞其它的消息,一定会造成消息积压。

    RocketMQ和Kafka在发送消息的时候都可以指定一个queue/partitions(发消息的时候指定一个key,通过key的hash找一个queue,相同的key得到的就是同一个queue)。

    1. RocketMQ 通过顺序消息对queue加锁变成单线程消费
    2. Kafka 的每一个partitions 就只有一个线程去消费

    消息可能重复消费这个和顺序消息没关系,所以在写消费逻辑的时候应该做幂等。

  • 相关阅读:
    最强分布式搜索引擎——ElasticSearch
    Java 之 CAS 原理及实现是怎样的?
    爬虫使用什么库更事半功倍?
    类与对象(八)----重载OverLoad
    第59节——redux-toolkit中的createSelector
    麦克风阵列入门
    Vite知识体系
    vue报错sockjs-node/info?t=或者报错info?t=
    SpringBoot学习(三)——yaml语法
    Linux系统struct input_event结构体分类型(鼠标、键盘、触屏)详解与例子
  • 原文地址:https://blog.csdn.net/Tomwildboar/article/details/136596943