钦州市网站建设_网站建设公司_改版升级_seo优化

一、延迟队列

一、概念 && 应用场景

延迟队列（Delayed Queue）即消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

延迟队列的使用场景有很多，比如：

1. 智能家居：用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延迟队列，当指令设定的时间到了再将指令推送到智能设备。

2. 日常管理：预定会议后，需要在会议开始前十五分钟提醒参会人参加会议。

3. 用户注册成功后，7天后发送短信，提高用户活跃度等。

4. …

RabbitMQ 本身没有直接支持延迟队列的功能，但是可以通过TTL+死信队列的组合模拟出延迟队列的功能，所以死信队列章节展示的也是延迟队列的使用。

假设一个应用中需要将每条消息都设置为 10 秒的延迟，生产者通过normal_exchange这个交换器将发送的消息存储在normal_queue这个队列中。消费者订阅的并非是normal_queue这个队列，而是dl_queue死信队列。当消息从normal_queue这个队列中过期之后被存入dl_queue这个队列中，消费者就恰巧消费到了延迟 10 秒的这条消息。

二、TTL+死信队列实现

延迟队列，就是希望等待特定的时间之后，消费者才能拿到这个消息。TTL 刚好可以让消息延迟一段时间成为死信，成为死信的消息会被投递到死信队列里，这样消费者一直消费死信队列里的消息就可以了。

1. 声明以及配置队列：（沿用前面死信队列的配置，只不过略做修改）

   @Bean("normalQueue")publicQueuenormalQueue(){returnQueueBuilder.durable(Constants.NORMAL_QUEUE).deadLetterExchange(Constants.DL_EXCHANGE)// 绑定死信交换机.deadLetterRoutingKey("dlk")// 绑定死信路由键.build();}

2. 发送消息：发送两条消息，一条消息 10s 后过期，第二条 20s 后过期

   @RequestMapping("/delay")publicStringdelay(){// 发送两条单独带TTL的消息rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test 10s..."+newDate(),message->{message.getMessageProperties().setExpiration("10000");// 延迟10s到达死信队列returnmessage;});rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test 20s..."+newDate(),message->{message.getMessageProperties().setExpiration("20000");// 延迟20s到达死信队列returnmessage;});return"发送成功！";}

3. 消费者：监听死信队列，打印信息，观察现象

   // 监听死信队列@RabbitListener(queues=Constants.DL_QUEUE)publicvoiddlQueue(Messagemessage){System.out.printf("%tc 死信队列接收到消息: %s\n",newDate(),newString(message.getBody()));}

该实现方式存在的问题🐔

把生产消息的顺序修改一下：先发送 20s 过期数据，再发送 10s 过期数据：

@RequestMapping("/delay")publicStringdelay(){// 发送两条单独带TTL的消息rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test 20s..."+newDate(),message->{message.getMessageProperties().setExpiration("20000");// 延迟20s到达死信队列returnmessage;});rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test 10s..."+newDate(),message->{message.getMessageProperties().setExpiration("10000");// 延迟10s到达死信队列returnmessage;});return"发送成功！";}

这时会发现：10s 过期的消息在 20s 后才进入到死信队列？？

这是因为消息过期之后，不一定会被马上丢弃。因为 RabbitMQ 只会检查队首消息是否过期，如果过期则丢到死信队列，此时就会造成一个问题，如果第一个消息的延时时间很长，第二个消息的延时时间很短，那第二个消息并不会优先得到执行。

所以在考虑使用TTL+死信队列实现延迟任务队列的时候，需要确认业务上每个任务的延迟时间是一致的，如果遇到不同的任务类型需要不同的延迟的话，需要为每一种不同延迟时间的消息建立单独的消息队列。

三、延迟队列插件

RabbitMQ 官方提供了一个延迟的插件来实现延迟的功能

参考：https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq

① 安装延迟队列插件

1. 下载并上传插件：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

根据自己的 RabbitMQ 版本选择相应版本的延迟插件，下载后上传到服务器或者放到本地的 RabbitMQ 的 plugins 目录中，可以参考下图解释：

2. 启动插件（下面是 linux 系统指令，其它系统指令直接问 gpt 即可）

   # 查看插件列表rabbitmq-plugins list# 启动插件rabbitmq-pluginsenablerabbitmq_delayed_message_exchange# 重启服务servicerabbitmq-server restart

3. 验证插件

在 RabbitMQ 管理平台查看，新建交换机时是否有延迟消息选项，如果有就说明延迟消息插件已经正常运行了。

② 基于插件延迟队列实现

1. 声明与绑定交换机、队列

   // 常量publicstaticfinalStringDELAY_EXCHANGE="delay_exchange";publicstaticfinalStringDELAY_QUEUE="delay_queue";// 延迟队列@Bean("delayQueue")publicQueuedelayQueue(){returnQueueBuilder.durable(Constants.DELAY_QUEUE).build();// 队列正常设置}@Bean("delayExchange")publicDirectExchangedelayExchange(){returnExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build();}@Bean("delayBinding")publicBindingdelayBinding(@Qualifier("delayQueue")Queuequeue,@Qualifier("delayExchange")Exchangeexchange){returnBindingBuilder.bind(queue).to(exchange).with("delay").noargs();}

2. 生产者发送两条消息，并设置延迟时间

   @RequestMapping("/delay")publicStringdelay(){// 发送两条单独带TTL的消息rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,"delay","delay test 10s..."+newDate(),message->{message.getMessageProperties().setDelayLong(20000L);// 延迟20s到达死信队列returnmessage;});rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,"delay","delay test 20s..."+newDate(),message->{message.getMessageProperties().setDelayLong(10000L);// 延迟10s到达死信队列returnmessage;});return"发送成功！";}

3. 消费者监听延迟队列，打印并观察消息

   @ComponentpublicclassDelayListener{@RabbitListener(queues=Constants.DELAY_QUEUE)publicvoiddelayQueue(Messagemessage){System.out.printf("%tc 延迟队列接收到消息: %s\n",newDate(),newString(message.getBody()));}}

从结果可以看出，使用延迟队列，可以保证消息按照延迟时间到达消费者。

四、两种实现方式的区别

二、事务

RabbitMQ 是基于 AMQP 协议实现的，该协议实现了事务机制，因此 RabbitMQ 也支持事务机制。Spring AMQP 也提供了对事务相关的操作。

RabbitMQ 事务允许开发者确保消息的发送和接收是原子性的，要么全部成功，要么全部失败。

要使用 RabbitMQ 事务，需要同时完成下面三步操作！

一、配置事务管理器

因为需要配置事务管理器，所以通常单独配置RabbitTemplate，然后配置时候调用rabbitTemplate.setChannelTransacted(true)打开事务管理器，并且需要配置一下事务管理器RabbitTransactionManager，如下所示：

@ConfigurationpublicclassTransactionConfig{@BeanpublicRabbitTransactionManagertransactionManager(ConnectionFactoryconnectionFactory){returnnewRabbitTransactionManager(connectionFactory);}@Bean("transRabbitTemplate")publicRabbitTemplatetransRabbitTemplate(ConnectionFactoryconnectionFactory){RabbitTemplaterabbitTemplate=newRabbitTemplate(connectionFactory);rabbitTemplate.setChannelTransacted(true);// 开启事务returnrabbitTemplate;}}

二、声明队列

声明队列就和普通队列一样，不需要什么特殊设置：

// 事务@Bean("transQueue")publicQueuetransQueue(){returnQueueBuilder.durable("transQueue").build();}

三、发送消息时打开事务

@Resource(name="transRabbitTemplate")privateRabbitTemplatetransRabbitTemplate;// 注入transRabbitTemplate@Transactional@RequestMapping("/trans")publicStringtrans(){transRabbitTemplate.convertAndSend("","transQueue","test trans 1...");inta=5/0;// 模拟出现异常transRabbitTemplate.convertAndSend("","transQueue","test trans 2...");return"发送成功！";}

如果三个步骤中没做其中的任何一个，都没办法保证事务机制的启动！（自行测试）

三、消息分发

一、概念

当 RabbitMQ 队列拥有多个消费者时，队列会把收到的消息分派给不同的消费者。每条消息只会发送给订阅列表里的一个消费者（普通队列的点对点消费）。这种方式非常适合扩展，如果现在负载加重，那么只需要创建更多的消费者来消费处理消息即可。

默认情况下，RabbitMQ 是以轮询的方法进行分发的，而不管消费者是否已经消费并已经确认了消息。这种方式是不太合理的，试想一下，如果某些消费者消费速度慢，而某些消费者消费速度快，就可能会导致某些消费者消息积压，某些消费者空闲，进而应用整体的吞吐量下降。

如何解决❓❓❓

可以使用channel.basicQos(int prefetchCount)，限制当前信道上的消费者所能保持的最大未确认消息的数量。

其中参数prefetchCount设置为 0 时表示没有上限。

比如：消费端调用了channel.basicQos(5)，RabbitMQ 会为该消费者计数，发送一条消息计数+1，消费一条消息计数-1，当达到了设定的上限，RabbitMQ 就不会再向它发送消息了，直到消费者确认了某条消息。类似 TCP/IP 中的 “滑动窗口”。

💥注意事项：basicQos()对拉模式的消费无效。

二、应用场景

消息分发的常见应用场景有如下：

1. 限流
2. 非公平分发

① 限流

如下场景：

订单系统每秒最多处理 5000 个请求，正常情况下，订单系统可以正常满足需求。

但是在秒杀时间点，请求瞬间增多，每秒 1w 个请求，如果这些请求全部通过 MQ 发送到订单系统，无疑会把订单系统压垮。

所以 RabbitMQ 提供了限流机制，可以控制消费端一次只拉取 N 个请求，保证消费端的正常运行。

操作：设置prefetchCount参数，同时设置消息确认机制为手动应答manual。

1. 配置prefetch参数，设置应答方式为手动应答

   spring:rabbitmq:addresses:amqp://liren:123123@127.0.0.1/lirendadalistener:simple:acknowledge-mode:manual# 手动确认prefetch:5

2. 配置交换机，队列

   // 常量publicstaticfinalStringQOS_EXCHANGE="qos_exchange";publicstaticfinalStringQOS_QUEUE="qos_queue";// 消息分发@Bean("qosQueue")publicQueueqosQueue(){returnQueueBuilder.durable(Constants.QOS_QUEUE).build();}@Bean("qosExchange")publicDirectExchangeqosExchange(){returnExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();}@Bean("qosBinding")publicBindingqosBinding(@Qualifier("qosQueue")Queuequeue,@Qualifier("qosExchange")Exchangeexchange){returnBindingBuilder.bind(queue).to(exchange).with("qos").noargs();}

3. 发送消息，一次发送20条消息

   @RequestMapping("/qos")publicStringqos(){for(inti=0;i<20;++i){rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE,"qos","test qos..."+i);}return"发送成功！";}

4. 消费者监听，进行手动确认

   @ConfigurationpublicclassQosListener{@RabbitListener(queues=Constants.QOS_QUEUE)publicvoidqosQueue(Messagemessage,Channelchannel)throwsIOException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();System.out.printf("接收到消息：%s，deliveryTag：%d%n",newString(message.getBody()),deliveryTag);// channel.basicAck(deliveryTag, true); // 注释掉，不进行确认，观察现象}}

发送消息时，需要先把手动确认注释掉，不然会直接消费掉

将prefetch注释掉，然后重新启动程序观察现象：

可以看到消息一次性都被消费者拿到了，就没有限流效果了！

② 负载均衡

如下图所示，在有两个消费者的情况下，一个消费者处理任务非常快，另一个非常慢，就会造成一个消费者会一直很忙，而另一个消费者很闲。这是因为 RabbitMQ 只是在消息进入队列时分派消息，它不考虑消费者未确认消息的数量。

我们可以使用设置prefetch=1的方式，告诉 RabbitMQ 一次只给一个消费者一条消息，也就是说，在处理并确认前一条消息之前，不要向该消费者发送新消息。此时，它会将它分派给下一个不忙的消费者。

1. 配置prefetch参数，设置应答方式为手动应答

   spring:rabbitmq:addresses:amqp://liren:123123@127.0.0.1/lirendadalistener:simple:acknowledge-mode:manual# 手动确认消息prefetch:1

2. 启动两个消费者（用休眠模拟业务处理耗时的不同）

   @ConfigurationpublicclassQosListener{@RabbitListener(queues=Constants.QOS_QUEUE)publicvoidqosQueue1(Messagemessage,Channelchannel)throwsIOException,InterruptedException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();System.out.printf("qosQueue1 接收到消息：%s，deliveryTag：%d%n",newString(message.getBody()),deliveryTag);Thread.sleep(1000);// 模拟快业务处理，1schannel.basicAck(deliveryTag,true);}@RabbitListener(queues=Constants.QOS_QUEUE)publicvoidqosQueue2(Messagemessage,Channelchannel)throwsIOException,InterruptedException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();System.out.printf("qosQueue2 接收到消息：%s，deliveryTag：%d%n",newString(message.getBody()),deliveryTag);Thread.sleep(2000);// 模拟慢业务处理，2schannel.basicAck(deliveryTag,true);}}

💥注意：deliveryTag有重复是因为两个消费者使用的是不同的Channel，每个Channel上的deliveryTag是独立计数的。