消费端如何保证消息队列MQ的有序消费

消息无序产生的原因

消息队列,既然是队列就能保证消息在进入队列,以及出队列的时候保证消息的有序性,显然这是在消息的生产端(Producer),但是往往在生产环境中有多个消息的消费端(Consumer),尽管消费端在拉取消息时是有序的,但各个消息由于网络等方面原因无法保证在各个消费端中处理时有序。

场景分析

先后两次修改了商品信息,消息A和消息B先后同步写入MySQL,接着异步写入消息队列中发送消息,此时消息队列生产端(Producer)按时序先后发出了A和B两条消息(消息A先发出,消息B后发出)。按业务逻辑,商品信息的最终状态需要以消息A和消息B综合为准。

看似一个比较常见的同步写数据库,异步发送消息的场景,但实际上需要保证消息的有序消费。

  • 假设1:消息A只包含修改的商品名称,消息B只包含修改的商品重量,此时消息队列的消费端实际上不需要关注消息时序,消息队列消费端(Consumer)只管消费即可。
  • 假设2:消息A包含修改的商品名称、重量,消息B包含修改的商品名称,此时消费端首先接收到消息B,后接收到消息A,那么消息B的修改就会被覆盖。此时消息队列的消费端实际上又需要关注消息时序

可见,你无法保证消息中包含什么信息,此时必须保证消息的有序消费。

业务角度如何保证消息有序消费

  • 生产端在发送消息时,始终保证消息是全量信息。
  • 消费端在接收消息时,通过缓存时间戳的方式,消费消息时判断消息产生的时间是否最新,如果不是则丢弃,如果是则执行下一步。

下面通过伪代码的方式描述:

生产端伪代码

insertWare(ware); #插入数据到数据库,通常在插入数据库时我们只会update修改的字段,而不会全量插入

ware = selectWareById(ware.getId); #获取商品的全量信息(此时是最新的),用于将它放入到消息队列中

syncMq(ware); #异步发送mq消息A

消费端伪代码

ware = fetchWare(); #获取消息

if (isLasted(ware)) #通过商品的修改时间戳判断是否是最新的修改

? TODO #执行下一步业务逻辑

else

? return #丢弃该消息

重点在于消费端如何判断该消息是否是最新的修改也就是isLasted方法。

isLasted方法

Long modified = getCacheById(ware.getId); #获取缓存中该条商品的最新修改时间

If (ware.getModified > modified) { #如果消息中商品修改时间大于缓存中的时间,说明是最新操作

? setCacheById(ware); #将该条消息的商品修改时间戳写入到缓存中

? return true;
} else #如果消息中的商品修改时间小于缓存中的时间,说明该条消息属于“历史操作”,不对其更新

? return false;

以上就是通过伪代码的方式,描述如何通过业务手段保证消息有序消费,重点在于全量发送信息和缓存时间戳。在其中还有一些技术实现细节。

例如:消费端消费消息B,执行到获取时间戳缓存之后,并在重新设置新的缓存之前,此时另一个消费端恰好也正在消费B它也正执行到获取时间戳缓存,由于消息A此时并没有更新缓存,消息A拿到的缓存仍然是旧的缓存,这时就会存在两个消费端都认为自己所消费的消息时最新的,造成该丢弃的消息没丢。

显然,这是分布式线程安全问题,分布式锁通常使用Redis或者ZooKeeper,加锁后的执行时序如下图所示。

这是从业务角度保证消息在消费端有序消费。通过在消息发送端全量发送消息以及在消息消费端缓存时间戳就可以保证消息的有序消费。

在上述场景中是先同步写入MySQL,再获取商品全量数据,接着再异步发送消息。这一系列的步骤可以通过接MySQL的binlog实现,在同步写入MySQL后,MySQL发送binlog变更,通过阿里巴巴Canal中间件接收MySQL的binlog变更再发送消息到消息队列。

原文地址:https://blog.51cto.com/14230003/2424783

时间: 2024-08-30 06:23:17

消费端如何保证消息队列MQ的有序消费的相关文章

RabbitMQ消息丢失问题和保证消息可靠性-消费端不丢消息和HA(二)

继续上篇文章解决RabbitMQ消息丢失问题和保证消息可靠性(一) 未完成部分,我们聊聊MQ Server端的高可用和消费端如何保证消息不丢的问题? 回归上篇的内容,我们知道消息从生产端到服务端,为了保证消息不丢,我们必须做哪些事情? 发送端采用Confirm模式,注意Server端没成功通知发送端,需要重发操作需要额外处理 消息的持久化处理 上面两个操作保证消息到服务端不丢,但是非高可用状态,如果节点挂掉,服务暂时不可用,需要重启后,消息恢复,消息不会丢失,因为有磁盘存储. 本文先从消费端讲起

转:为什么会需要消息队列(MQ)?

为什么会需要消息队列(MQ)? ########################################################################################## 主要原因是由于在高并发环境下,由于来不及同步处理,请求往往会发生堵塞,比如说,大量的insert,update之类的请求同时到达MySQL,直接导致无数的行锁表锁,甚至最后请求会堆积过多,从而触发too many connections错误.通过使用消息队列,我们可以异步处理请求,从而

消息队列一:为什么需要消息队列(MQ)?

为什么会需要消息队列(MQ)? ########################################################################################## 主要原因是由于在高并发环境下,由于来不及同步处理,请求往往会发生堵塞,比如说,大量的insert,update之类的请求同时到达MySQL,直接导致无数的行锁表锁,甚至最后请求会堆积过多,从而触发too many connections错误.通过使用消息队列,我们可以异步处理请求,从而

C#消息队列(MQ)零基础从入门到实战演练

一.课程介绍 如果您从工作中之听过但未有接触过消息对队列(MQ),如果你接触过一点关于MQ的知识,如果没有这么的多如果的话......,那么阿笨将通过本次<C#消息队列零基础从入门到实战演练>分享课让您对消息队列有一个实质性的了解和认识,达到实际的灵活贯通和运用.本次分享课您将学习到以下知识点: 1.微软MSMQ的基本使用技能以及MSMQ在WCF技术中的运用. 2.企业级RabbitMQ消息队列的两种消费模式(生产消费和发布订阅)的介绍和使用. 3.如何实现RabbitMQ客户端(Client

消息队列MQ

消息队列MQ MQ全称为Message Queue 消息队列(MQ)是一种应用程序对应用程序的通信方法.MQ是消费-生产者模型的一个典型的代表,一端往消息队列中不断写入消息,而另一端则可以读取队列中的消息.这样发布者和使用者都不用知道对方的存在. 生产消费模型: ''' 生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题.生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列

关于MQ的几件小事(二)如何保证消息队列的高可用

1.RabbitMQ的高可用 RabbitMQ基于主从模式实现高可用.RabbitMQ有三种模式:单机模式,普通集群模式,镜像集群模式. (1)单机模式: 单机模式就是demo级别的,生产中不会有人使用. (2)普通集群模式 普通集群模式就是在多台机器上启动多个rabbitmq实例,每个机器启动一个.但是创建的queue只会放在一个rabbitmq实例上面,但是其他的实例都同步了这个queue的元数据.在你消费的时候,如果连接到了另一个实例,他会从拥有queue的那个实例获取消息然后再返回给你.

消息队列(MQ)

1. 分类: 获取消息方式:A. push(推)方式:优点——可以尽可能快地将消息发送给消费者,缺点——如果消费者处理能力跟不上,消费者的缓冲区可能会溢出:     B. pull(拉)方式:优点——消费端可以按处理能力进行拉去,缺点——会增加消息延迟: 能否重复消费:A. 点对点(Queue):不可重复消费: B. 发布/订阅(Topic):可重复消费. 2. 特点: A. 先进先出(FIFO): B. 发布订阅: C. 持久化: D. 分布式. 3. 技术组件 A. RabbitMQ:是使用

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

考虑一个分布式场景中一个常见的场景:服务A执行某个数据库操作成功后,会发送一条消息到消息队列,现在希望只有数据库操作执行成功才发送这条消息.下面是一些常见的作法: 1. 先执行数据库操作,再发送消息 public void purchaseOrder() { orderDao.save(order); messageQueue.send(message); } 有可能order新增成功,发送消息失败.最终形成不一致状态. 2. 先发送消息,再执行数据库操作 public void purchas

C#编写Windows服务程序 (服务端),client使用 消息队列 实现淘宝 订单全链路效果

需求: 针对 淘宝提出的 订单全链路 产品接入 .http://open.taobao.com/doc/detail.htm?id=102423&qq-pf-to=pcqq.group oms(订单管理系统) 实现  , 完毕后 效果:在千牛工作台 --订单全链路  可看到效果例如以下图   -------------------------------------------------------------------------------------------------------