首页 > 其他分享 >MQ系列12:如何保证消息顺序性

MQ系列12:如何保证消息顺序性

时间:2023-06-07 14:57:30浏览次数:38  
标签:12 队列 顺序 MQ 消息 有序性 消费

MQ系列1:消息中间件执行原理
MQ系列2:消息中间件的技术选型
MQ系列3:RocketMQ 架构分析
MQ系列4:NameServer 原理解析
MQ系列5:RocketMQ消息的发送模式
MQ系列6:消息的消费
MQ系列7:消息通信,追求极致性能
MQ系列8:数据存储,消息队列的高可用保障
MQ系列9:高可用架构分析
MQ系列10:如何保证消息幂等性消费
MQ系列11:如何保证消息可靠性传输

1 介绍

消息的有序性在很多业务场景中占有很重要的位置。
比如购物场景,需要按照 创建订单 --> 订单付款 --> 完成订单 顺序执行。
又比如出行场景,接单 --> 接送到达目的地 --> 付款 --> 完成订单。
这种是严格按照顺序执行的,这样的顺序消费才不会出问题,而且各个订单之间是互相独立和并行执行的。
所以,在MQ中,如何稳定地保证顺序性消息处理,是一个不可避免的话题。

2 消息的有序性说明

消息的有序执行,一般不是单个组件的能力。而是整个消息从生产,排队,存储到消费都是有序的,比如上面提到的购物和出行场景。
这就要求我们在消息队列(如果是Kafka,还是RocketMQ、RabbitMQ)中,保证以下前提:

  • 消息生产的有序性:即生产者组件有序发送消息
  • 消息入出队列的有序性:即消息是按照进入的先后顺序排队列放的,遵循FIFO原则。
  • 消息的存储的有序性:与上一点一致,部分场景下为了提高可用,就是要持久化到磁盘,这时候应该遵循有序存放,才能保证后续有序消费
  • 消息消费的有序性:即按照顺序进行消费。又分为全局顺序消息与部分顺序消息,全局是指Topic下的所有消息都要保证顺序;部分顺序消息保证每一组消息被顺序消费即可。

这边还有个问题,如果想让全局都是顺序性消费,那么只能用一个消费者去消费队列(一般来说也是单个生产者),这是会严重影响整体性能的,一般没这个,都是分组顺序执行消费的。
image

2.1 消息生产的有序性

要保证整个消息队列的有序性执行,首先要保证消息生产的有序性。
RocketMQ在Broker中防止了很多Topic,主题(Topic)可以看做消息的归类,我们将消息进行类型划分,相同类型的消息称为一个 Topic。比如我们在淘宝或京东上购买商品的的过程,就可能产生:购物车消息、交易消息、物流消息等,1条消息必然归属于1个 Topic 。
1个 Topic可以有0 ~ n 个生产者向其发送消息;也可以被 0~n 个消费者订阅和处理,于是就有出现了生产者组和消费者组,如下图:
image

或者同一个Topic中,创建不同的Queue,同一个消息生产者将消息隔离发送到不同的Queue中:
image

按照上述的模式,同理,我们只需要保证一组相同的消息按照给定的顺序存入同一个队列中,就能保证生产者有序存储,比如一次完整的消费过程:创建订单、付款、完成订单按照顺序在一个队列(Queue)中执行那就可以了。

★ 同时我们要保证同一组的消息在消息生产的时候投送到一个组中。这个相对来说不难,可以这么做:

  • 比如一个订单的多个子消息的父订单号是一致,我们把这些消息按照订单号取模,投送到对应的Queue中就行了,比如 订单号 % 队列数量( 163105015 % 9)
  • 发送消息自定义消息标签(消息标签可以用队列编号命名),一组消息使用同一个标签,改组标签对应的消息都投向标签所在的队列。

★ 业务程序方面,必须使用同步发送的方式,这样才能保证生产者发送的消息有序,否则按照FIFO的原则,很可能 订单完成 会被先消费。
但是我们业务程序,比如Java代码中为了提升性能,可能使用多线程的模式进行事件触发。多线程下保证生产者顺序性,可以使用锁并配合 spring的publish event(按照顺序执行的内部队列),持久化之后,再按照先进先出的顺序推送消息进入MQ中。
可以参考下 ,大概就是将你的事件进行顺序化一下。

★ 上述方法也不能完完全全的避免顺序化执行。如果broker服务发生故障,或者消息发生丢失,都有可能导致事件消费不完整,出现不一致的问题。

2.2 消息有序性存储

Broker 存储架构采用文件存储机制(类似Kafka),即直接在磁盘上使用文件来保存消息,而不是采用Redis或者MySQL之类的持久化工具。
它会把消息存储所属相关的文件存储在ROCKETMQ_HOME下,包含三个部分:

  • CommitLog 消息元数据
  • ConsumeQueue 消息逻辑队列
  • IndexFile 索引文件

存储消息的元数据,所有消息都会顺序存入到CommitLog文件中。
ConsumeQueue是指存储消息在CommitLog上的索引,一个MessageQueue一个文件,记录当前MessageQueue被哪些消费者组消费到了哪一条CommitLog。
所以一切都是顺序性操作下来的,而且按照 MessageQueue 做了隔离了,不用担心乱序的问题。详细参考 《MQ系列8:数据存储,消息队列的高可用保障

image

2.3 消息消费的有序性

最后一步就是消费的有序性了,既然消息生产和消息持久化都可以做到有序性。那么只要保证消费的有序性,就能保证整个消息队列的有序执行。
这边以RocketMQ为例子,RockerMQ采用MessageListener 回调函数进行监听,监听到消息之后进行数据处理。MessageListener主要提供了两种消费模式,如下:

  • 有序消费模式MessageListenerOrderly
  • 并发消费模式MessageListenerConcurrently

其中有序消费模式有序消费模式MessageListenerOrderly可以保证按照顺序进行消息处理。但是消费的业务代码实现是多线程并行的,依然是无法保证的。
实际上RocketMQ也是这么做的,MessageListenerConcurrently拉到消息之后会提交到线程池去消费,而MessageListenerOrderly则是通过分布式锁和本地锁保证同时只有一条线程去消费一个队列(Queue)上的数据。
这种消费模式就是使用以下3把锁来确保顺序性:

  • broker端的分布式锁
  • messageQueue的本地synchronized锁
  • ProcessQueue的本地consumeLock

3 总结

要消息的顺序性消费:需要保持先后顺序的消息放到同一个消息队列中(kafka中就是partition,rabbitMq中就是queue),然后使用线程池消费的时候使用分布式锁和本地锁保证同时只有一条线程去消费一个队列(Queue)上的数据。

标签:12,队列,顺序,MQ,消息,有序性,消费
From: https://www.cnblogs.com/wzh2010/p/15888528.html

相关文章

  • Golang中如何控制goroutine的执行顺序?
    首先说明一下原理:前后协程之间通过通道去相互限制,后一个线程尝试去获取一个channel的值,当channel中没有值时,就会一直阻塞,而前一个协程则负责关闭channel,当前一个协程完成了这个操作,后一个协程才可以结束阻塞,继续执行。示例代码:packagemainimport( "fmt" "time")funcma......
  • 文件读取顺序差异
    Linux(Ubuntu):Windows:......
  • YS9082HC+B27B固件量产工具,YS9082HT可参考,YS9082HC+镁光MT29F512G08EBLCE开卡!YS9082HP
    YS9082HC+B27B,镁光MT29F512G08EBLCE开卡!闪存ID:2C,C3,08,32,E6,00。如下图,不知道为什么检测出来的是9081?开卡设置,从量产部落下载的YS9082HCMPTool,如下图:结果报错:重新设置,更改了大小,240G改到了160G!分析是坏块过多了!我有不少B27颗粒的坏块都多,还是主控问题?我的其他两片B27,开120G都......
  • 12万短文学网句子内容大全ACCESS数据库
    闲来无事,看到“短文学网”文章内容还算整洁,而且非常容易进行采集,于是也就手痒了弄了一下,速度非常快可能与网络没有大量广告啊、JS啊有关。详细的分类信息如下:qq日志包含有:qq空间(2098)条、非主流日(180)条、搞笑日志(132)条、个性日志(204)条、经典日志(260)条、空间文字(848)......
  • SX1262 与 SX1278、SX1276 有什么区别?
    通常的物联网解决方案和设备一直都非常昂贵,或在实施中不切合实际。理想的无线连接技术应该是低成本、高可靠性的,可进行长距离传输,且拥有超长的电池续航时间。像zigbee、Bluetooth和Wi-Fi这样的短距离技术不能满足这些要求,蜂窝移动网络虽然支持更长的距离,但电池续航时间太短。为了......
  • RabbitMQ 集群环境搭建
    1.使用集群的原因最开始我们介绍了如何安装及运行RabbitMQ服务,不过这些是单机版的,无法满足目前真实应用的要求。如果RabbitMQ服务器遇到内存崩溃、机器掉电或者主板故障等情况,该怎么办?单台RabbitMQ服务器可以满足每秒1000条消息的吞吐量,那么如......
  • spring boot 集成 Neo4j org.neo4j.ogm.metadata.DomainInfo.useClassgraph(DomainIn
    springboot版本:2.2.13.RELEASE 问题在于引入后,报错spring-boot-starter-data-neo4j<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-neo4j</artifactId></dependency>  *......
  • 7.12 字符串查找
    containsindexOf,lastIndexOf,startsWith,endWithpublicclassHelloWorld{publicstaticvoidmain(Stringargs[]){//Stringargs[]字符串数组的意思Stringstr="www.mldn.cn";System.out.println(str.contains("mldn"));//tru......
  • rocketmq
    rocketmq在RocketMQ中,msgId和offsetMsgId是两个不同的概念。msgId是消息唯一标识符。具有全局唯一性,由RocketMQ在消息发送时自动生成。当消息被成功发送到Broker端后,msgId会随着消息一起存储在commitlog文件中,并且在消息消费时也可以用来查找和定位该消息。通常情......
  • Spring 学习笔记(12)—— AOP 基础
    AOP到底是什么AOP术语创建增强类......