work模式

RabbitMQ是一种开源的消息队列软件它实现了高级消息队列协议AMQP提供了可靠的消息传递机制以及支持分布式应用程序之间的通信。

RabbitMQ支持多种编程语言如Java、Python、Ruby、PHP等等并且可以在不同的操作系统上运行如Windows、Linux、Mac

RabbitMQ的核心理念是分离应用程序之间的通信允许开发人员将其应用程序解耦从而使它们更加容易理解、扩展和维护。

在使用RabbitMQ时开发人员可以将消息发布到队列中当消费者连接到队列时RabbitMQ会自动将消息传递给消费者。

此外RabbitMQ还提供了一些高级功能如重试机制、消息优先级、发布/订阅模式等等使得开发人员可以更加灵活地控制消息队列的行为。

rabbitmq安装图文保姆级教程

mode也称为点对点模式是最简单的模式。

生产者将消息发送到队列中消费者从队列中读取消息并处理。

这种模式只有一个生产者和一个消费者消息被传递一次并且只被一个消费者接收生产者和消费者之间的关系是一对一的

代码示例

新建父级项目,再在父级项目中新建consumer和producter两个子模块

两个子级添加必须依赖

dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactIdversion2.6.13/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependency

添加mq配置

可以看到生产者每生产一条消息,消费者都可以监听到消息并且进行消费

work模式

mode也称为竞争消费者模式多个消费者同时订阅同一个队列中的消息但是只有一个消费者可以消费每个消息。

这种模式用于负载均衡其中每个消费者处理的消息数量相等或者最接近相等并且每个消息只被处理一次。

代码示例

我们这里用生产者发送20条消息,建立两个消费者来进行消息的监听,且消费者1每次消费消息前设置睡眠1秒,消费者2每次设置睡眠2秒,然后观察消费者的消费情况

work生产者:

可以看到消费者1和消费者2都是消费了10条消息,也就是说无论消费者1和消费者2谁消费消息的速度快慢,mq默认都是平均分配的,这样很明显不太符合能者多劳的情况

添加消费者配置

spring.rabbitmq.listener.simple.acknowledge-modemanual

#设置预取数量为1

spring.rabbitmq.listener.simple.prefetch1

重启测试:

此时可以明显看到速度更快的消费者1消费了13条,消费者2只消费了7条,已经达到了能者多劳的预期

pubsub模式

mode一个生产者将消息发送到交换机Exchange中该交换机将消息传递给多个队列所有与这些队列绑定的消费者都会接收到该消息。

这种模式用于广播消息。

代码示例

mode生产者将消息发送到交换机中并将消息标记为一个特定的路由键Routing

Key。

交换机将消息传递给绑定到该交换机的与该路由键匹配的队列。

这种模式用于将消息路由到特定的队列中。

代码示例

mode也称为通配符模式它是路由模式的升级版。

生产者将消息发送到交换机中并将消息标记为一个主题Topic交换机将消息传递给绑定到该交换机的与该主题匹配的队列。

这种模式与路由模式类似但是主题模式的主题可以使用通配符允许复杂的路由规则。

生产者

下面以用户在下单付款后系统为用户添加积分的案例为例来对mq进行场景使用

用户积分表

通过orderNo进行订单付款,生产者生成付款消息,付款后用户订单状态调整为已付款状态,消费者消费该消息并且为用户积分表添加在原有积分基础上添加该订单所奖励的积分,积分值这里直接取订单表中的productPrice字段数据

初始化数据库表

付款订单号为1336557511364313088订单,status付款状态初始化为0(未付款),1是付款状态

该订单目前积分为

积分新增成功,由原先的1000,新增了订单的productPrice的1399变为了2399

消费成功

注意在启动测试这个案例前先把之前测试时设置的手动签收和预取关掉,否则消费者不会自动实时消费生产者生产的消息

可靠性投递是指在消息队列中确保消息得到正确且可靠地传递即使在出现网络故障或服务器宕机等异常情况下消息也不会丢失。

在RabbitMQ中可靠性投递通常采用以下两个概念

生产者确认Publisher

Confirms当生产者将消息发送到RabbitMQ时会等待RabbitMQ发送确认消息告诉生产者消息已被成功接收。

如果RabbitMQ没有收到消息或者消息发送失败则会通知生产者重新发送消息。

消费者确认Consumer

Acknowledgements当消费者从队列中接收消息时它们会发送一个确认消息给RabbitMQ告诉RabbitMQ已经接收到并处理了该消息。

如果消费者未发送确认消息则RabbitMQ会认为消息未被正确处理从而重新将消息发送给消费者。

生产者服务中开启配置

spring.rabbitmq.publisher-confirm-typecorrelated

当消息经生产者投递到交换机后为避免消息丢失需要回调RabbitTemplate.ConfirmCallback接口回调接口后尤其是要对投递失败的消息进行处理或者记录下来保证消息不丢失。

该接口不管消息投递到交换机成功或者失败都会进行回调未避免消息丢失可以选择在回调接口中只处理或者登记投递失败的消息达到消息不丢失的目的。

交换机投递确认回调

先在mq图像化管理中找到一个真实存在的交换机进行成功投递,查看成功投递后的回调

测试投递

上面成功投递的示例中还存在一定问题,那就是交换机投递成功是后面的队列是否投递成功没有检测到,之前测试的队列是rtt,但是图型化中是没有这个队列的

保证消息在发送队列处不丢失

消息由交换机和消息队列中异常导致消息丢失问题解决办法就是在添加消息从交换机路由到队列中失败后回调的接口在回调接口中把失败的消息保存下来就可以避免消息丢失了。

添加发布确认返回

spring.rabbitmq.publisher-returnstrue

追加实现returnCallback

创建一个具有延时消息处理能力的队列并将队列绑定到延时消息交换机上。

此时需要指定交换机的

delayed_routing_key

属性值自动进行延迟发送。

当延迟时间到达后消息会被投递到绑定的队列中进行处理。

延时消息的实现可以确保消息在指定时间后被投递从而提高了消息系统的可靠性和稳定性。

原理图

注意这里消费者监听的是delay_queue2,这是一个正常的消费队列,不是死信队列,死信队列可以理解为没有消费者监听的队列,且消息有过期时间

前面在图形化中新建死信队列时已经设置过过期时间为10000ms(10s)

启动测试:

可以看到生产者生产了几条消息后,在图形化中的delay_queue1中有消息堆叠,10秒内不会被消费掉,10秒后消费者日志才陆续进行消费打印,图形化中的消息也被消费掉不再堆叠

如果这里看的清晰,可以先将消费者停掉,直接启动生产者,看看10秒过期后消息会不会从delay_queue1自动转发到delay_queue2

可以看到消息过期后自动转发到delay_queue2了,而delay_queue2之前消费者是正常监听的,所以由delay_queue1的消息过期时间间接就实现了延迟消息的效果

纯代码创建

死信队列是一个特殊的队列用于存储无法被消费的消息这些消息通常被定义为无法被路由或者由于一些原因被拒绝。

死信队列通常会在原始消息被拒绝、超时、过期或达到最大重试次数等情况下被触发。

队列过期时间指的是队列本身的过期时间在该队列中的所有消息的过期时间都是一致的,在该时间到达时队列会自动删除。

而消息过期时间指的是消息本身的过期时间在该时间到达时消息会被标记为过期然后会被发送到死信队列。

两者的作用不同队列过期时间控制整个队列的生命周期而消息过期时间只控制单个消息的生命周期。

当消息过期时间配置到队列中时当消息在队列中等待时如果消息已经达到过期时间那么这个消息将会被视为死信消息并被发送到死信队列中。

如果队列本身已经到达过期时间则这个队列将会被删除而其中所有的消息也会成为死信消息并被发送到死信队列中。

因此两者的区别在于队列过期时间会影响所有消息而消息过期时间只影响每个消息的生命周期。

新建配置类创建交换机和队列并进行绑定

org.springframework.amqp.core.Binding;

import

org.springframework.amqp.core.BindingBuilder;

import

org.springframework.amqp.core.DirectExchange;

import

org.springframework.amqp.core.Queue;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

DirectExchange(ttl_queue_exchange,true,

false);}Beanpublic

过期时间,队列容纳消息数等*/MapString,Object

map

单位毫秒map.put(x-message-ttl,15000);//x-max-length

固定属性

表示当前队列中最多可以存放的消息条数map.put(x-max-length,5);return

new

false,false,map);}//消息队列和交换机进行绑定Beanpublic

Binding

BindingBuilder.bind(ttlQueue()).to(ttlQueueExchange()).with(ttl_key);}

}图形化界面中现在还有没有ttl_queue_exchange交换机和ttl_queue队列

生产者接口,之前设置队列时设置队列最多存储5条消息,过期时间15秒,这里直接生产10条消息进行测试

启动生产者进行测试:

可以看到生产者虽然生产了10条消息,但是队列中始终只存了5条消息,且15秒后队列中的所有消息自动过期,此时没有消费者服务启动,说明死信队列已生效

生产者接口

可以看到该消息30秒后过期拿不到消息体了,单个消息过期设置成功

当生产者的消息投递到队列后,由于消息设置的有过期时间.在指定时间内没有被消费者消费,此时队列中的消息就会成为死信队列

创建一个正常监听队列的交换机,一个没有消费者监听的过期时间队列,再创建一个监听的死信交换机和死信队列

生产者接口

启动测试,注意此时不启动消费者服务,只观察消息会不会由ttl_time_queue队列转投到dead_queue队列:

可以看到生产者在ttl_time_queue中投递一条消息后,由于没有消费者消费,该队列的消息在设置的过期时间20秒后自动转投给了dead_exchange交换机,交换机又投给了dead_letter_queue队列,由于此时没有消费者所以消息也在dead_letter_queue中没有消费,这里为了方便观察没有开启消费者,下面再启动一次有消费者版的直接观察最后死信队列的延时消费效果

启动测试

可以看到这次ttl_time_queue过期的消息直接被监听dead_letter_queue的消费者消费了,从而间接实现了延时消费

消息溢出

当生产者投递到队列的消息数量超出队列容量时,超出的队列也会被投入到死信队列中

生产者接口,模拟生产10条消息

观察现象,注意此时也不启动消费者服务,只观察消息会不会由max_queue队列转投到dead_queue队列

可以看到生产者投递了10条消息,有3条是进入到了max_queue中,剩下溢出的7条都转投到了dead_queue队列

此时再启动消费者重新投递10条消息,由于此时max_queue中已经有了3条消息,此时再投递的消息就都是溢出消息查看多出的溢出死信消息是否直接被消费掉

可以看到溢出的消息变成死信消息直接被消费掉了,从而也实现了死信队列

消息被拒

当生产者的消息投递到队列后被消费者拒绝,此时队列中的消息也会成为死信

创建reject_queue队列并配置消息拒绝后转投的交换机和路由

拒绝接收消息

启动测试,先观察reject_queue队列被先消费者拒绝消费后消息是否自动转投到死信队列dead_letter_queue中:

可以看到监听reject_queue的消费者拒绝了该队列的消息后自动将消息转投到dead_letter_queue中了

下面将监听

dead_letter_queue的消费者放开,直接测试消息拒绝的死信队列,观察其现象

启动测试:

可以看到监听reject_queue的消费者拒绝了该队列的消息后自动将消息转投到dead_letter_queue中后直接被监听dead_letter_queue的消费者消费掉了,从而也实现了死信队列

import

org.springframework.amqp.core.Binding;

import

org.springframework.amqp.core.BindingBuilder;

import

org.springframework.amqp.core.DirectExchange;

import

org.springframework.amqp.core.Queue;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

DirectExchange(ttl_queue_exchange,true,

false);}Beanpublic

过期时间,队列容纳消息数等*/MapString,Object

map

单位毫秒map.put(x-message-ttl,15000);//x-max-length

固定属性

表示当前队列中最多可以存放的消息条数map.put(x-max-length,5);return

new

false,false,map);}//消息队列和交换机进行绑定Beanpublic

Binding

BindingBuilder.bind(ttlQueue()).to(ttlQueueExchange()).with(ttl_key);}//

DirectExchange

DirectExchange(ttl_msg_exchange,true,

false);}Beanpublic

BindingBuilder.bind(ttlMsgQueue()).to(ttlMsgExchange()).with(ttl_msg_key);}//监听消息的交换机Beanpublic

DirectExchange

DirectExchange(listen_exchange,true,

DirectExchange

DirectExchange(dead_exchange,true,

Queue

BindingBuilder.bind(deadLetterQueue()).to(deadExchange()).with(dead_key);}//时间过期队列Beanpublic

Queue

ttlTimeQueue(){MapString,Object

map

HashMap();//指定给当前被拒绝消费的队列配置拒绝后消息转发的死信交换机//x-dead-letter-exchange

固定属性

表示指定的死信交换机map.put(x-dead-letter-exchange,

dead_exchange);//

设置死信交换机绑定的队列之间的路由键map.put(x-dead-letter-routing-key,

dead_key);map.put(x-message-ttl,20000);//

new

BindingBuilder.bind(ttlTimeQueue()).to(listenExchange()).with(ttl_key);}//规定数量队列Beanpublic

Queue

HashMap();//指定给当前被拒绝消费的队列配置拒绝后消息转发的死信交换机//x-dead-letter-exchange

固定属性

表示指定的死信交换机map.put(x-dead-letter-exchange,

dead_exchange);//

设置死信交换机绑定的队列之间的路由键map.put(x-dead-letter-routing-key,

dead_key);map.put(x-max-length,3);//

设置队列最大容量数return

BindingBuilder.bind(maxQueue()).to(listenExchange()).with(max_key);}//消息拒绝

队列,绑定配置Beanpublic

HashMap();//指定给当前被拒绝消费的队列配置拒绝后消息转发的死信交换机//x-dead-letter-exchange

固定属性

表示指定的死信交换机map.put(x-dead-letter-exchange,

dead_exchange);//

设置死信交换机绑定的队列之间的路由键map.put(x-dead-letter-routing-key,

dead_key);return

BindingBuilder.bind(rejectQueue()).to(listenExchange()).with(reject_key);}}死信队列踩坑

在实际操作中博主也遇到了一些比较容易出现问题的坑,也在这里记录下

前面我们使用死信队列成功实现了延迟消息,但是在正常消费者里是需要填写监听的正常消费队列的

如果这里不填写队列,会造成消息一直堆叠在delay_queue2中,原先博主以为这里交换机和路由都已经绑定过了,可以不填写就行,但是事实还是狠狠打了一巴掌,把错误地方也复现下

我们把这里的队列名称去掉,其他的生产者等都不动,再启动测试看看会有什么现象

启动测试:

可以看到消费者虽然正常打印了消费日志,但是在图形化队列中队列还是一直在堆叠,博主以为只是图像化出了问题,但是把队列名重新加上后再重启消费者,消费者又重新打印消费了刚才的几条消息,而此时图像化界面队列中也不再堆叠消息了,真是大坑一个

此处如果填了队列名,代码中不再填写路由key也是可以正常消费的,看示例

删除路由key,重新测试

可以看到路由key在图形化界面中已经绑定了,这里不用再指定也可以正常消费

rabbitmq实战场景源码