大家好，我是小❤，一个漂泊江湖多年的 985 非科班程序员，曾混迹于国企、互联网大厂和创业公司的后台开发攻城狮。
在今天的文章中，我们来聊一聊 RabbitMQ，这是小 ❤ 在工作中用的最早的消息中间件，主要用于大量数据的异步消费。
RabbitMQ 是一个开源的消息中间件，它实现了高级消息队列协议（AMQP），同时提供了各种重要组件来支持消息的生产、传输和消费。
图片
RabbitMQ 的工作方式是基于生产者、交换机和队列之间的协作。这是一个简单的消息传递过程：
这种模型具有高度的灵活性，可以轻松处理大量消息，同时确保消息的可靠传递。
说到消息中间件，很多人首先想到的就是 Kafka，但 RabbitMQ 也是许多金融或互联网公司构建可靠、可伸缩和高性能系统的首选。
这是为什么呢？
主要得从 RabbitMQ 的特性说起，主要有二：一个是功能强大，另一个是可靠性！
RabbitMQ 注重消息的可靠性和灵活性，适合任务排队和消息传递。而 Kafka 是分布式流式平台，注重日志存储和数据分发。
顺序消费也是可靠性的一种，RabbitMQ 可以使用单一队列或多个单一队列来确保顺序消费。
除此之外，RabbitMQ 还提供持久性队列和消息，以确保消息在 RabbitMQ 服务器宕机后不会丢失。另外，生产者可以使用发布确认机制来确认消息是否被接收。
RabbitMQ 相对 kafka 可靠性更好，数据更不易丢失，这对于一些数据敏感型的业务来说，显然更适合用前者。
并且，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"] = 10ch.Publish( "tizi365", // 交换机 "", // 路由参数 false, false, amqp.Publishing{ Priority:5, // 设置消息优先级 DeliveryMode:2, // 消息投递模式，1代表非持久化，2代表持久化， ContentType: "text/plain", Body: []byte(body), })
当优先级队列消费生效时，会首先消费高优先级队列中的优先级高的消息，以此来实现顺序消费。
但需要注意的是，优先级队列触发的条件比较苛刻，在需要严格保证业务消息顺序的情况下最好不要使用！
RabbitMQ 里，当消息在队列中变成死信（消费者无法正常处理的消息）之后，它会被重新投递到一个交换机上（即死信交换机），死信交换机上绑定的消费队列就是死信队列。
图片
死信产生需要满足如下条件：
当死信产生时，如果我们定义了一个死信交换机（其实就是一个普通的交换机，只是用于处理死信，所以叫死信交换机），然后在死信交换机上绑定了一个队列（称作死信队列）。
最后，如果死信队列有消费者监听时，死信消息的处理就会和正常业务消息一样，从交换机到队列，再由死信消费者（监听死信队列的消费者）正常消费。
RabbitMQ 本身不支持延时队列，但是我们可以通过 RabbitMQ 的插件 rabbitmq-delayed-message-exchange，或者使用 死信队列 + 消息过期 的方式来实现。
当我们在电商里购物，或者在 12306 买票时，大概都会遇到这样一个场景：每次下订单后，到支付订单中间有一段商品锁定时间，超过时间后未支付订单就会关闭。
状态转换图如下：
图片
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), })
该方案的核心思想是，先创建死信交换机、队列和消费者，来监听死信消息。
然后创建定时过期的消息，比如订单支付的时间为 30min，则将消息的 TTL（最大存活时间）设置为 30min，将消息放到一个没有消费者消费的队列中，当消息过期后就会成为死信。
死信消息被重新发送到死信交换机，然后我们在死信队列中消费该消息，根据商品 ID 判断该商品是否被支付。
如果没有支付，就取消订单，修改订单状态为待下单。如果已经支付，就将商品状态修改为已完成，并丢掉这条死信消息。
RabbitMQ 是一个功能强大的消息中间件，它在许多互联网应用中扮演了关键角色，比如华为摄像机 SDK 的监控图像数据上报，大部分电商系统的异步消费等等。
本文链接：http://www.28at.com/showinfo-26-16867-0.html深入浅出RabbitMQ：顺序消费、死信队列和延时队列
声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com
Copyright © 2016-2023 天津谷骐科技有限公司 版权所有 sitemap.xml
违法及侵权请联系：2376512515@qq.com 津ICP备18001702号
津公网安备 12010102000574号
