您当前的位置:首页 > 电脑百科 > 程序开发 > 编程百科

深入浅出RabbitMQ:顺序消费、死信队列和延时队列

时间:2023-11-03 14:00:25  来源:微信公众号  作者:xin猿意码

1. 引言

在今天的文章中,我们来聊一聊 RabbitMQ,这是小 ❤ 在工作中用的最早的消息中间件,主要用于大量数据的异步消费。

2. RabbitMQ

2.1 核心组件

RabbitMQ 是一个开源的消息中间件,它实现了高级消息队列协议(AMQP),同时提供了各种重要组件来支持消息的生产、传输和消费。

深入浅出RabbitMQ:顺序消费、死信队列和延时队列图片

  • Producer(生产者): 生产者是消息的发送方,负责将消息发布到 RabbitMQ 服务器。消息可以包含任何内容,例如任务、日志、通知等。
  • Channel(信道):消息推送与接收时使用的通道。
  • Exchange(交换机): 交换机是消息的中转站,它接收来自生产者的消息并将其路由到一个或多个队列。不同类型的交换机,如 fanout,direct,topic,headers,支持不同的路由规则。
  • Queue(队列): 队列是消息的缓冲区,消息在发送到消费者之前存储在队列中,消费者从队列中获取消息并进行处理。
  • Consumer(消费者): 消费者是消息的接收方,它从队列中获取消息并进行处理。消费者可以是多个,它们可以在不同的应用程序或服务器上运行。

2.2 工作流程

RabbitMQ 的工作方式是基于生产者、交换机和队列之间的协作。这是一个简单的消息传递过程:

  • 将队列与交换机绑定(Binding)起来,定义了消息的路由规则;
  • 生产者将消息发布到交换机,交换机根据绑定规则将消息路由到一个或多个队列;
  • 消费者从队列中获取消息并进行处理。

这种模型具有高度的灵活性,可以轻松处理大量消息,同时确保消息的可靠传递。

2.3 特性

说到消息中间件,很多人首先想到的就是 Kafka,但 RabbitMQ 也是许多金融互联网公司构建可靠、可伸缩和高性能系统的首选。

这是为什么呢?

主要得从 RabbitMQ 的特性说起,主要有二:一个是功能强大,另一个是可靠性!

RabbitMQ 注重消息的可靠性和灵活性,适合任务排队和消息传递。而 Kafka 是分布式流式平台,注重日志存储和数据分发。

顺序消费也是可靠性的一种,RabbitMQ 可以使用单一队列或多个单一队列来确保顺序消费。

除此之外,RabbitMQ 还提供持久性队列和消息,以确保消息在 RabbitMQ 服务器宕机后不会丢失。另外,生产者可以使用发布确认机制来确认消息是否被接收。

RabbitMQ 相对 kafka 可靠性更好,数据更不易丢失,这对于一些数据敏感型的业务来说,显然更适合用前者。

并且,RabbitMQ 中原生支持死信队列,可以更好地处理未完成的业务消息,以及实现延时队列等特性,接下来我们一一介绍。

3. 保证顺序消费

RabbitMQ 提供了多个队列模型来保证消息的顺序消费。这对于某些应用程序非常重要,例如处理订单、支付和库存管理。

消息错乱消费的场景

深入浅出RabbitMQ:顺序消费、死信队列和延时队列图片

如上图所示,有三条业务消息分别是删除、增加和修改操作,但是 Consumer 没有按顺序消费,最终存储的顺序是增加、修改和删除,就会发生数据错乱。

针对消息有序性的问题,RabbitMQ 的解决方法是分三个阶段来保证。

  • 发送消息:入队列

消息发送时,需要业务来保证顺序性,就是保证生产者入队的顺序是有序的。

在分布式的场景下如果难以保证各个服务器的入队顺序,则可以加分布式锁的方式来解决。或者在业务生产方的消息里带上消息递增 ID,以及消息产生的时间戳。

  • 队列中的消息

在 RabbitMQ 的消息会保存在队列(Queue)中,在同一个队列里的消息是先进先出(FIFO)的,这个由 RabbitMQ 来帮我们保证顺序。

而不同队列中的消息,RabbitMQ 无法保证其顺序性,就像我们在食堂打饭一样,站在不同的排队队列,我们也无法保证会比其他队列的人先打上饭。

  • 消费消息:出队

一般来说,出队后的顺序消费交给消费者去保证。我们说的保证消费顺序,通常也是指消费者消费消息的顺序。

有多个消费者的情况下,通常是无法保证消息顺序的。

这就相当于我们在排队打饭时,有多个打饭阿姨,但是每个阿姨打饭的速度不一致,对应我们消费者的消费能力也不同。

所以,为了保证消息的顺序性,我们可以只使用一个消费者来接收业务消息。

就好比只有一个阿姨在打饭,来得早就一定能早点打上饭。但很明显,这样效率不是很高,所以在使用时我们需要权衡利弊:看业务更需要顺序性,还是更需要消费效率。

优先级队列

在保证顺序消费时,另一个迂回策略是可以使用优先级队列(Priority Queue)。

在 RabbitMQ3.5 之后,当消费者数量较少,如果服务器检测到消费者不能及时消费消息的情况下,优先级队列就会生效。

具体有两种优先级策略:

  • 设置队列的优先级
  • 设置消息的优先级

在声明队列时,我们可以通过 x-max-priority 属性来设置队列的最大优先级,或通过 Priority 属性来设置消息的优先级,从 1~10。

Golang 实现代码如下:

// 队列属性
props := make(map[string]interface{})
// 设置队列最大优先级
props["x-max-priority"] = 10

ch.Publish(
   "tizi365",     // 交换机
   "", // 路由参数
   false,
   false,
   amqp.Publishing{
       Priority:5, // 设置消息优先级
       DeliveryMode:2,  // 消息投递模式,1代表非持久化,2代表持久化,
       ContentType: "text/plAIn",
       Body:       []byte(body),
  })

当优先级队列消费生效时,会首先消费高优先级队列中的优先级高的消息,以此来实现顺序消费。

但需要注意的是,优先级队列触发的条件比较苛刻,在需要严格保证业务消息顺序的情况下最好不要使用!

4. 死信队列

RabbitMQ 里,当消息在队列中变成死信(消费者无法正常处理的消息)之后,它会被重新投递到一个交换机上(即死信交换机),死信交换机上绑定的消费队列就是死信队列。

深入浅出RabbitMQ:顺序消费、死信队列和延时队列图片

死信的产生

死信产生需要满足如下条件:

  • 消息被消费者手动拒绝接收,并且 requeue(重新加入队列)策略为 False;
  • 消息已经过期(TTL);
  • 队列达到最大长度,消息装不下了。

死信的处理步骤

当死信产生时,如果我们定义了一个死信交换机(其实就是一个普通的交换机,只是用于处理死信,所以叫死信交换机),然后在死信交换机上绑定了一个队列(称作死信队列)。

最后,如果死信队列有消费者监听时,死信消息的处理就会和正常业务消息一样,从交换机到队列,再由死信消费者(监听死信队列的消费者)正常消费。

5. 延时队列

RabbitMQ 本身不支持延时队列,但是我们可以通过 RabbitMQ 的插件 rabbitmq-delayed-message-exchange,或者使用 死信队列 + 消息过期 的方式来实现。

5.1 应用场景

当我们在电商里购物,或者在 12306 买票时,大概都会遇到这样一个场景:每次下订单后,到支付订单中间有一段商品锁定时间,超过时间后未支付订单就会关闭。

状态转换图如下:

深入浅出RabbitMQ:顺序消费、死信队列和延时队列图片

5.2 插件实现

  • 安装插件

Github 地址:

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

从 github 的 release 页面的 assets, 下载 rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez 文件,把文件放到 rabbitmq 插件目录(plugins目录)

提示:版本号可能跟本教程不一样,如果你的 rabbitmq 就是最新版本,插件也选择最新版本就行。

  • 激活插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  • 定义交换机

通过 x-delayed-type 设置自定义交换机属性,支持发送延迟消息:

props := make(map[string]interface{})
   //关键参数,支持发送延迟消息
   props["x-delayed-type"] = "direct"

   // 声明交换机
   err = ch.ExchangeDeclare(
       "delay.queue",   // 交换机名字
       "fanout", // 交换机类型
       true,     // 是否持久化
       false,    
       false,
       false,
       props,      // 设置属性
  )
  • 发送延迟消息

通过消息头(x-delay),设置消息延迟时间。

msgHeaders := make(map[string]interface{})
       // 通过消息头,设置消息延迟时间,单位毫秒
       msgHeaders["x-delay"] = 6000

       err = ch.Publish(
           "delay.queue",     // 交换机名字
           "", // 路由参数
           false,
           false,
           amqp.Publishing{
               Headers:msgHeaders, // 设置消息头
               ContentType: "text/plain",
               Body:       []byte(body),
          })

5.3 死信队列 + 消息过期方案

该方案的核心思想是,先创建死信交换机、队列和消费者,来监听死信消息。

然后创建定时过期的消息,比如订单支付的时间为 30min,则将消息的 TTL(最大存活时间)设置为 30min,将消息放到一个没有消费者消费的队列中,当消息过期后就会成为死信。

死信消息被重新发送到死信交换机,然后我们在死信队列中消费该消息,根据商品 ID 判断该商品是否被支付。

如果没有支付,就取消订单,修改订单状态为待下单。如果已经支付,就将商品状态修改为已完成,并丢掉这条死信消息。

5. 小结

RabbitMQ 是一个功能强大的消息中间件,它在许多互联网应用中扮演了关键角色,比如华为摄像机 SDK 的监控图像数据上报,大部分电商系统的异步消费等等。



Tags:RabbitMQ   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,不构成投资建议。投资者据此操作,风险自担。如有任何标注错误或版权侵犯请与我们联系,我们将及时更正、删除。
▌相关推荐
RabbitMQ如何实现延迟队列?
延迟队列是指当消息被发送以后,并不是立即执行,而是等待特定的时间后,消费者才会执行该消息。延迟队列的使用场景有以下几种: 未按时支付的订单,30 分钟过期之后取消订单。 给活...【详细内容】
2024-01-26  Search: RabbitMQ  点击:(46)  评论:(0)  加入收藏
RabbitMQ消息顺序性解密:保证消息的正确顺序
在分布式系统中,保证消息的正确顺序对于一些应用场景至关重要。而RabbitMQ作为一种流行的消息队列系统,本身并不提供严格的消息顺序保证。下面将探讨如何在使用RabbitMQ时实现...【详细内容】
2023-12-04  Search: RabbitMQ  点击:(126)  评论:(0)  加入收藏
RabbitMQ与消息限流策略的完美结合
在当今互联网时代,高并发访问已成为许多应用系统面临的常见挑战之一。对于需要处理大量请求的系统来说,如何保证系统的稳定性和可靠性是一个关键问题。RabbitMQ作为一种可靠的...【详细内容】
2023-11-27  Search: RabbitMQ  点击:(165)  评论:(0)  加入收藏
实时协作的秘诀:RabbitMQ与WebSockets的结合
实时协作是现代软件开发中非常重要的一个方面。为了实现实时协作,一种常见的做法是将消息队列与WebSocket技术相结合。其中,RabbitMQ是一个功能强大的消息队列系统,它能够有效...【详细内容】
2023-11-21  Search: RabbitMQ  点击:(176)  评论:(0)  加入收藏
RabbitMQ中的消息持久化策略与存储优化实践
本文将介绍RabbitMQ中的消息持久化策略,并提供一些存储优化的实践方法,帮助您确保消息的可靠性和系统的性能。在RabbitMQ消息队列中,消息的可靠性传输和持久化是非常重要的。下...【详细内容】
2023-11-15  Search: RabbitMQ  点击:(241)  评论:(0)  加入收藏
Centos7下安装部署RabbitMQ,看这篇就够了
前言RabbitMQ是一个开源的强大的企业消息系统,支持主流的操作系统,支持多种开发语言。我们项目中使用RabbitMQ作为消息队列,解耦业务,构建高可靠的消息队列系统。RabbitMQ可以...【详细内容】
2023-11-09  Search: RabbitMQ  点击:(306)  评论:(0)  加入收藏
RabbitMQ发送和接收消息的几种方式
channel.basicQos(0, 1, false):0表示对消息的大小无限制,1表示每次只允许消费一条,false表示该限制不作用于channel。同时,我们采用手工ACK的方式,因为我们配置文件配置了 spri...【详细内容】
2023-11-08  Search: RabbitMQ  点击:(261)  评论:(0)  加入收藏
RabbitMQ的四种交换机详解
交换机主要是接收消息并且转发到绑定的队列,交换机不存储消息,在启用ack模式后,交换机找不到队列会返回错误。交换机有四种类型:Direct, topic, Headers and Fanout。图片一、to...【详细内容】
2023-11-06  Search: RabbitMQ  点击:(264)  评论:(0)  加入收藏
深入浅出RabbitMQ:顺序消费、死信队列和延时队列
1. 引言在今天的文章中,我们来聊一聊 RabbitMQ,这是小 ❤ 在工作中用的最早的消息中间件,主要用于大量数据的异步消费。2. RabbitMQ2.1 核心组件RabbitMQ 是一个开源的消息中间...【详细内容】
2023-11-03  Search: RabbitMQ  点击:(178)  评论:(0)  加入收藏
在Linux系统中实现容器化的消息中间件:RabbitMQ和Kafka
消息中间件在现代分布式系统中起着至关重要的作用。它们可以在不同的应用程序之间实现可靠的异步通信,提供高吞吐量、低延迟和可扩展性。下面将介绍如何在Linux系统中使用容...【详细内容】
2023-09-08  Search: RabbitMQ  点击:(374)  评论:(0)  加入收藏
▌简易百科推荐
即将过时的 5 种软件开发技能!
作者 | Eran Yahav编译 | 言征出品 | 51CTO技术栈(微信号:blog51cto) 时至今日,AI编码工具已经进化到足够强大了吗?这未必好回答,但从2023 年 Stack Overflow 上的调查数据来看,44%...【详细内容】
2024-04-03    51CTO  Tags:软件开发   点击:(5)  评论:(0)  加入收藏
跳转链接代码怎么写?
在网页开发中,跳转链接是一项常见的功能。然而,对于非技术人员来说,编写跳转链接代码可能会显得有些困难。不用担心!我们可以借助外链平台来简化操作,即使没有编程经验,也能轻松实...【详细内容】
2024-03-27  蓝色天纪    Tags:跳转链接   点击:(12)  评论:(0)  加入收藏
中台亡了,问题到底出在哪里?
曾几何时,中台一度被当做“变革灵药”,嫁接在“前台作战单元”和“后台资源部门”之间,实现企业各业务线的“打通”和全域业务能力集成,提高开发和服务效率。但在中台如火如荼之...【详细内容】
2024-03-27  dbaplus社群    Tags:中台   点击:(8)  评论:(0)  加入收藏
员工写了个比删库更可怕的Bug!
想必大家都听说过删库跑路吧,我之前一直把它当一个段子来看。可万万没想到,就在昨天,我们公司的某位员工,竟然写了一个比删库更可怕的 Bug!给大家分享一下(不是公开处刑),希望朋友们...【详细内容】
2024-03-26  dbaplus社群    Tags:Bug   点击:(5)  评论:(0)  加入收藏
我们一起聊聊什么是正向代理和反向代理
从字面意思上看,代理就是代替处理的意思,一个对象有能力代替另一个对象处理某一件事。代理,这个词在我们的日常生活中也不陌生,比如在购物、旅游等场景中,我们经常会委托别人代替...【详细内容】
2024-03-26  萤火架构  微信公众号  Tags:正向代理   点击:(10)  评论:(0)  加入收藏
看一遍就理解:IO模型详解
前言大家好,我是程序员田螺。今天我们一起来学习IO模型。在本文开始前呢,先问问大家几个问题哈~什么是IO呢?什么是阻塞非阻塞IO?什么是同步异步IO?什么是IO多路复用?select/epoll...【详细内容】
2024-03-26  捡田螺的小男孩  微信公众号  Tags:IO模型   点击:(8)  评论:(0)  加入收藏
为什么都说 HashMap 是线程不安全的?
做Java开发的人,应该都用过 HashMap 这种集合。今天就和大家来聊聊,为什么 HashMap 是线程不安全的。1.HashMap 数据结构简单来说,HashMap 基于哈希表实现。它使用键的哈希码来...【详细内容】
2024-03-22  Java技术指北  微信公众号  Tags:HashMap   点击:(11)  评论:(0)  加入收藏
如何从头开始编写LoRA代码,这有一份教程
选自 lightning.ai作者:Sebastian Raschka机器之心编译编辑:陈萍作者表示:在各种有效的 LLM 微调方法中,LoRA 仍然是他的首选。LoRA(Low-Rank Adaptation)作为一种用于微调 LLM(大...【详细内容】
2024-03-21  机器之心Pro    Tags:LoRA   点击:(12)  评论:(0)  加入收藏
这样搭建日志中心,传统的ELK就扔了吧!
最近客户有个新需求,就是想查看网站的访问情况。由于网站没有做google的统计和百度的统计,所以访问情况,只能通过日志查看,通过脚本的形式给客户导出也不太实际,给客户写个简单的...【详细内容】
2024-03-20  dbaplus社群    Tags:日志   点击:(4)  评论:(0)  加入收藏
Kubernetes 究竟有没有 LTS?
从一个有趣的问题引出很多人都在关注的 Kubernetes LTS 的问题。有趣的问题2019 年,一个名为 apiserver LoopbackClient Server cert expired after 1 year[1] 的 issue 中提...【详细内容】
2024-03-15  云原生散修  微信公众号  Tags:Kubernetes   点击:(6)  评论:(0)  加入收藏
站内最新
站内热门
站内头条