聊一聊顺序消息(RocketMQ顺序消息的实现机制)

当我们说顺序时,我们在说什么?

日常思维中,顺序大部分情况会和时间关联起来,即时间的先后表示事件的顺序关系。

比如事件A发生在下午3点一刻,而事件B发生在下午4点,那么我们认为事件A发生在事件B之前,他们的顺序关系为先A后B。

上面的例子之所以成立是因为他们有相同的参考系,即他们的时间是对应的同一个物理时钟的时间。如果A发生的时间是北京时间,而B依赖的时间是东京时间,那么先A后B的顺序关系还成立吗?

如果没有一个绝对的时间参考,那么A和B之间还有顺序吗,或者说怎么断定A和B的顺序?

显而易见的,如果A、B两个事件之间如果是有因果关系的,那么A一定发生在B之前(前因后果,有因才有果)。相反,在没有一个绝对的时间的参考的情况下,若A、B之间没有因果关系,那么A、B之间就没有顺序关系。

那么,我们在说顺序时,其实说的是:

  • 有绝对时间参考的情况下,事件的发生时间的关系;
  • 和没有时间参考下的,一种由因果关系推断出来的happening before的关系;

在分布式环境中讨论顺序

当把顺序放到分布式环境(多线程、多进程都可以认为是一个分布式的环境)中去讨论时:

  • 同一线程上的事件顺序是确定的,可以认为他们有相同的时间作为参考
  • 不同线程间的顺序只能通过因果关系去推断

(点表示事件,波浪线箭头表示事件间的消息)

上图中,进程P中的事件顺序为p1->p2->p3->p4(时间推断)。而因为p1给进程Q的q2发了消息,那么p1一定在q2之前(因果推断)。但是无法确定p1和q1之间的顺序关系。

推荐阅读《Time, Clocks, and the Ordering of Events in a Distributed System》,会透彻的分析分布式系统中的顺序问题。

消息中间件中的顺序消息

什么是顺序消息

有了上述的基础之后,我们回到本篇文章的主题中,聊一聊消息中间件中的顺序消息。

顺序消息(FIFO 消息)是 MQ 提供的一种严格按照顺序进行发布和消费的消息类型。顺序消息由两个部分组成:顺序发布和顺序消费。

顺序消息包含两种类型:

分区顺序:一个Partition内所有的消息按照先进先出的顺序进行发布和消费

全局顺序:一个Topic内所有的消息按照先进先出的顺序进行发布和消费

这是阿里云上对顺序消息的定义,把顺序消息拆分成了顺序发布和顺序消费。那么多线程中发送消息算不算顺序发布?

如上一部分介绍的,多线程中若没有因果关系则没有顺序。那么用户在多线程中去发消息就意味着用户不关心那些在不同线程中被发送的消息的顺序。即多线程发送的消息,不同线程间的消息不是顺序发布的,同一线程的消息是顺序发布的。这是需要用户自己去保障的。

而对于顺序消费,则需要保证哪些来自同一个发送线程的消息在消费时是按照相同的顺序被处理的(为什么不说他们应该在一个线程中被消费呢?)。

全局顺序其实是分区顺序的一个特例,即使Topic只有一个分区(以下不在讨论全局顺序,因为全局顺序将面临性能的问题,而且绝大多数场景都不需要全局顺序)。

如何保证顺序

在MQ的模型中,顺序需要由3个阶段去保障:

  1. 消息被发送时保持顺序
  2. 消息被存储时保持和发送的顺序一致
  3. 消息被消费时保持和存储的顺序一致

发送时保持顺序意味着对于有顺序要求的消息,用户应该在同一个线程中采用同步的方式发送。存储保持和发送的顺序一致则要求在同一线程中被发送出来的消息A和B,存储时在空间上A一定在B之前。而消费保持和存储一致则要求消息A、B到达Consumer之后必须按照先A后B的顺序被处理。

如下图所示:

对于两个订单的消息的原始数据:a1、b1、b2、a2、a3、b3(绝对时间下发生的顺序):

  • 在发送时,a订单的消息需要保持a1、a2、a3的顺序,b订单的消息也相同,但是a、b订单之间的消息没有顺序关系,这意味着a、b订单的消息可以在不同的线程中被发送出去
  • 在存储时,需要分别保证a、b订单的消息的顺序,但是a、b订单之间的消息的顺序可以不保证
    • a1、b1、b2、a2、a3、b3是可以接受的
    • a1、a2、b1、b2、a3、b3也是可以接受的
    • a1、a3、b1、b2、a2、b3是不能接受的
  • 消费时保证顺序的简单方式就是“什么都不做”,不对收到的消息的顺序进行调整,即只要一个分区的消息只由一个线程处理即可;当然,如果a、b在一个分区中,在收到消息后也可以将他们拆分到不同线程中处理,不过要权衡一下收益

开源RocketMQ中顺序的实现

上图是RocketMQ顺序消息原理的介绍,将不同订单的消息路由到不同的分区中。文档只是给出了Producer顺序的处理,Consumer消费时通过一个分区只能有一个线程消费的方式来保证消息顺序,具体实现如下。

Producer端

Producer端确保消息顺序唯一要做的事情就是将消息路由到特定的分区,在RocketMQ中,通过MessageQueueSelector来实现分区的选择。

  • List<MessageQueue> mqs:消息要发送的Topic下所有的分区
  • Message msg:消息对象
  • 额外的参数:用户可以传递自己的参数

比如如下实现就可以保证相同的订单的消息被路由到相同的分区:

long orderId = ((Order) object).getOrderId;
return mqs.get(orderId % mqs.size());

Consumer端

RocketMQ消费端有两种类型:MQPullConsumer和MQPushConsumer。

MQPullConsumer由用户控制线程,主动从服务端获取消息,每次获取到的是一个MessageQueue中的消息。PullResult中的List msgFoundList自然和存储顺序一致,用户需要再拿到这批消息后自己保证消费的顺序。

对于PushConsumer,由用户注册MessageListener来消费消息,在客户端中需要保证调用MessageListener时消息的顺序性。RocketMQ中的实现如下:

  1. PullMessageService单线程的从Broker获取消息
  2. PullMessageService将消息添加到ProcessQueue中(ProcessMessage是一个消息的缓存),之后提交一个消费任务到ConsumeMessageOrderService
  3. ConsumeMessageOrderService多线程执行,每个线程在消费消息时需要拿到MessageQueue的锁
  4. 拿到锁之后从ProcessQueue中获取消息

保证消费顺序的核心思想是:

  • 获取到消息后添加到ProcessQueue中,单线程执行,所以ProcessQueue中的消息是顺序的
  • 提交的消费任务时提交的是“对某个MQ进行一次消费”,这次消费请求是从ProcessQueue中获取消息消费,所以也是顺序的(无论哪个线程获取到锁,都是按照ProcessQueue中消息的顺序进行消费)

顺序和异常的关系

顺序消息需要Producer和Consumer都保证顺序。Producer需要保证消息被路由到正确的分区,消息需要保证每个分区的数据只有一个线程消息,那么就会有一些缺陷:

  • 发送顺序消息无法利用集群的Failover特性,因为不能更换MessageQueue进行重试
  • 因为发送的路由策略导致的热点问题,可能某一些MessageQueue的数据量特别大
  • 消费的并行读依赖于分区数量
  • 消费失败时无法跳过

不能更换MessageQueue重试就需要MessageQueue有自己的副本,通过Raft、Paxos之类的算法保证有可用的副本,或者通过其他高可用的存储设备来存储MessageQueue。

热点问题好像没有什么好的解决办法,只能通过拆分MessageQueue和优化路由方法来尽量均衡的将消息分配到不同的MessageQueue。

消费并行度理论上不会有太大问题,因为MessageQueue的数量可以调整。

消费失败的无法跳过是不可避免的,因为跳过可能导致后续的数据处理都是错误的。不过可以提供一些策略,由用户根据错误类型来决定是否跳过,并且提供重试队列之类的功能,在跳过之后用户可以在“其他”地方重新消费到这条消息。

原文地址:https://www.cnblogs.com/hzmark/p/orderly_message.html

时间: 2024-08-29 06:43:18

聊一聊顺序消息(RocketMQ顺序消息的实现机制)的相关文章

高可用保证消息绝对顺序消费的BROKER设计方案

转自: http://www.infoq.com/cn/articles/high-availability-broker-design?utm_source=tuicool&utm_medium=referral 在要求严格顺序消息的场景下,消息的发送者,BROKER端(BROKER端和消息存储放在一起),消息的消费者都要求按照顺序进行,三者任何一个环节的乱序都会导致消息最终的消费顺序被打乱. 如果为每一个消息维护一个有序的ID,发送和存储消息无序,消费逻辑会变得非常复杂,消费端要对消息进行重

Floodlight中 处理packetin消息的顺序(1)

当Controller和SW建立连接之后,就可以处理来自SW的各种OF msg.当接收到 packetin 消息之后,会将其分发给各个监听了这个OFMessage的listeners,所以如果我们要设计自己的控制器模块,只需要实现相应的接口方法,约定执行顺序即可.接口IListener 主要抽象了监听器模块的名字,执行顺序,接口IOFMessageListener则抽象了我们的Controller如何处理具体的这个openflow消息.这里通过阅读代码来判断这些模块处理packetin消息的相对

MFC窗口的消息响应顺序

最开始的消息传递 1.AfxWndProc() 该函数负责接收消息,找到消息所属的CWnd对象,然后调用AfxCallWndProc 2.AfxCallWndProc() 该函数负责保存消息(保存的内容主要是消息标识符和消息参数)供应用程序以后使用,然后调用WindowProc()函数 3.WindowProc() 该函数负责发送消息到OnWndMsg()函数,如果未被处理,则调用DefWindowProc()函数 4.OnWndMsg() 该函数的功能首先按字节对消息进行排序,对于WM_COM

MFC应用程序中处理消息的顺序

应用程序处理消息的过程: 1.AfxWndProc()        该函数负责接收消息,找到消息所属的CWnd对象,然后调用AfxCallWndProc 2.AfxCallWndProc()  该函数负责保存消息(保存的内容主要是消息标识符和消息参数)供应用程序以后使用,然后调用WindowProc()函数 3.WindowProc()       该函数负责发送消息到OnWndMsg()函数,如果未被处理,则调用DefWindowProc()函数 4.OnWndMsg()         该

MQ如何解决消息的顺序问题和消息的重复问题?

一.摘要 分布式消息系统作为实现分布式系统可扩展.可伸缩性的关键组件,需要具有高吞吐量.高可用等特点.而谈到消息系统的设计,就回避不了两个问题: 1.消息的顺序问题 2.消息的重复问题 二.关键特性以及其实现原理 2.1.顺序消息 要实现严格的顺序消息,简单且可行的办法就是: 保证生产者 - MQServer - 消费者是一对一对一的关系 这样的设计虽然简单易行,但也会存在一些很严重的问题,比如: 1.并行度就会成为消息系统的瓶颈(吞吐量不够) 2.更多的异常处理,比如:只要消费端出现问题,就会

如何保证消息的顺序性

1.面试官心里分析 其实这个也是用MQ的时候必问的话题,第一看看你了解不了解顺序这个事儿?第二看看你有没有办法保证消息是有顺序的?这个生产系统中常见的问题. 2.面试题剖析 我举个例子,我们以前做过一个mysql binlog同步的系统,压力还是非常大的,日同步数据要达到上亿.mysql -> mysql,常见的一点在于说大数据team,就需要同步一个mysql库过来,对公司的业务系统的数据做各种复杂的操作. 你在mysql里增删改一条数据,对应出来了增删改3条binlog,接着这三条binlo

阿里Java面试题剖析:在高并发的情况下如何保证消息的顺序性?

面试原题 如何保证消息的顺序性? 面试官心理分析 其实这个也是用 MQ 的时候必问的话题,第一看看你了不了解顺序这个事儿?第二看看你有没有办法保证消息是有顺序的?这是生产系统中常见的问题. 面试题剖析 我举个例子,我们以前做过一个 mysql binlog 同步的系统,压力还是非常大的,日同步数据要达到上亿,就是说数据从一个 mysql 库原封不动地同步到另一个 mysql 库里面去(mysql -> mysql).常见的一点在于说比如大数据 team,就需要同步一个 mysql 库过来,对公司

搞懂分布式技术19:使用RocketMQ事务消息解决分布式事务

搞懂分布式技术19:使用RocketMQ事务消息解决分布式事务 初步认识RocketMQ的核心模块 rocketmq模块 rocketmq-broker:接受生产者发来的消息并存储(通过调用rocketmq-store),消费者从这里取得消息. rocketmq-client:提供发送.接受消息的客户端API. rocketmq-namesrv:NameServer,类似于Zookeeper,这里保存着消息的TopicName,队列等运行时的元信息.(有点NameNode的味道) rocketm

RocketMQ事务性消息及持久化

TransactionProducer(事务消息): 在分布式系统中,我们时常会遇到分布式事务的问题,除了常规的解决方案之外,我们还可以利用RocketMQ的事务性消息来解决分布式事务的问题.RocketMQ和其他消息中间件最大的一个区别是支持了事务消息,这也是分布式事务里面的基于消息的最终一致性方案. RocketMQ消息的事务架构设计: 生产者执行本地事务,修改订单支付状态,并且提交事务 生产者发送事务消息到broker上,消息发送到broker上在没有确认之前,消息对于consumer是不