rabbitmq延遲隊(duì)列的使用方式
rabbitmq延遲隊(duì)列的使用
1、場(chǎng)景：
1.定時(shí)發(fā)布文章
2.秒殺之后，給30分鐘時(shí)間進(jìn)行支付，如果30分鐘后，沒有支付，訂單取消。
3.預(yù)約餐廳，提前半個(gè)小時(shí)發(fā)短信通知用戶。
A -> 13:00 17:00 16:30 延遲時(shí)間： 730 * 60 * 1000
B -> 11:00 18:00 17:30 延遲時(shí)間： 1330 * 60 * 1000
C -> 8:00 14:00 13:30 延遲時(shí)間： 11*30 * 60 * 1000
第一種方式：創(chuàng)建具有超時(shí)功能且綁定死信交換機(jī)的消息隊(duì)列
@Bean
public Queue directQueueLong(){
return QueueBuilder.durable("業(yè)務(wù)隊(duì)列名稱")
.deadLetterExchange("死信交換機(jī)名稱")
.deadLetterRoutingKey("死信隊(duì)列 RoutingKey")
.ttl(20000) // 消息停留時(shí)間
//.maxLength(500)
.build();
}監(jiān)聽死信隊(duì)列，即可處理超時(shí)的消息隊(duì)列
缺點(diǎn)：
上述實(shí)現(xiàn)方式中，ttl延時(shí)隊(duì)列中所有的消息超時(shí)時(shí)間都是一樣的，如果不同消息想設(shè)置不一樣的超時(shí)時(shí)間，就需要建立多個(gè)不同超時(shí)時(shí)間的消息隊(duì)列，比較麻煩，且不利于維護(hù)。
第二種方式：創(chuàng)建通用延時(shí)消息
rabbitTemplate.convertAndSend("交換機(jī)名稱", "RoutingKey","對(duì)象",
message => {
message.getMessageProperties().setExpiration(String.valueOf(5000))
return message;
}
);缺點(diǎn)：
該種方式可以創(chuàng)建一個(gè)承載不同超時(shí)時(shí)間消息的消息隊(duì)列，但是這種方式有一個(gè)問題，如果消息隊(duì)列中排在前面的消息沒有到超時(shí)時(shí)間，即使后面的消息到了超時(shí)時(shí)間，先到超時(shí)時(shí)間的消息也不會(huì)進(jìn)入死信隊(duì)列，而是先檢查排在最前面的消息隊(duì)列是否到了超時(shí)時(shí)間，如果到了超時(shí)時(shí)間才會(huì)繼續(xù)檢查后面的消息。
第三種方式：使用rabbitmq的延時(shí)隊(duì)列插件，實(shí)現(xiàn)同一個(gè)隊(duì)列中有多個(gè)不同超時(shí)時(shí)間的消息，并按時(shí)間超時(shí)順序出隊(duì)
1、下載延遲插件
在 RabbitMQ 的 3.5.7 版本之后，提供了一個(gè)插件（rabbitmq-delayed-message-exchange）來實(shí)現(xiàn)延遲隊(duì)列 ，同時(shí)需保證 Erlang/OPT 版本為 18.0 之后。
我這里 MQ 的版本是 3.10.0 現(xiàn)在去 GitHub 上根據(jù)版本號(hào)下載插件
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
2、安裝插件并啟用
我用的是 Docker 客戶端，下載完成后直接把插件放在 /opt/rabbitmq 目錄，然后拷貝到容器內(nèi)plugins目錄下（rabbitmq是容器的name，也可以使用容器id）
docker cp /opt/rabbitmq/rabbitmq_delayed_message_exchange-3.10.0.ez rabbitmq:/plugins
進(jìn)入 Docker 容器
docker exec -it rabbitmq /bin/bash
在plugins內(nèi)啟用插件
#先執(zhí)行，解除防火墻限制，增加文件權(quán)限 cd plugins umask 0022 rabbitmq-plugins enable rabbitmq_delayed_message_exchange
退出容器
exit
重啟 RabbitMQ
docker restart rabbitmq
通過UI查看
原理
代碼使用
消費(fèi)者
/*
* Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
*
*/
package com.fpl.consumers;
import cn.hutool.core.map.MapUtil;
import com.fpl.model.OrderingOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
/**
* <p>Project: spring-rabbitmq - DelayConsumer</p>
* <p>Powered by fpl1116 On 2024-04-09 11:34:07</p>
* <p>描述：<p>
*
* @author penglei
* @version 1.0
* @since 1.8
*/
@Configuration
@Slf4j
public class DelayConsumer {
@Bean
public Queue delayQueue1(){
return QueueBuilder.durable("Delay_Q01").lazy().build();
}
@Bean
public CustomExchange delayExchange(){
//參數(shù)x-delayed-type
Map<String, Object> map = MapUtil.of("x-delayed-type","direct");
return new CustomExchange("Delay_E01","x-delayed-message",true,false,map);
}
@Bean
public Binding binding1(Queue delayQueue1, CustomExchange delayExchange){
return BindingBuilder.bind(delayQueue1).to(delayExchange).with("RK01").noargs();
}
// @RabbitListener(queues = "Delay_Q01")
public void receiveMessage(OrderingOk msg) {
log.info("消費(fèi)者1 收到消息："+ msg );
}
}生產(chǎn)者
/*
* Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.
*
*/
package com.fpl.provider;
import com.fpl.model.OrderingOk;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* <p>Project: spring-rabbitmq - DelayProvider</p>
* <p>Powered by fpl1116 On 2024-04-09 11:35:51</p>
* <p>描述：<p>
*
* @author penglei
* @version 1.0
* @since 1.8
*/
@Service
public class DelayProvider {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send(OrderingOk orderingOk) {
rabbitTemplate.convertAndSend("Delay_E01", "RK01", orderingOk,new MessagePostProcessor(){
@Override
public Message postProcessMessage(Message message) throws AmqpException {
int id = orderingOk.getId();
int ttl = 0;
if(id == 1){
ttl = 50*1000;
}else if(id == 2){
ttl = 30*1000;
}else if(id ==3){
ttl = 40*1000;
}else if(id ==4){
ttl = 10*1000;
}else if(id ==5){
ttl = 20*1000;
}
//延遲交換機(jī)使用的delay參數(shù),設(shè)置消息的延期時(shí)長(zhǎng)，單位是微妙
message.getMessageProperties().setDelay(ttl);
//延遲交換機(jī)消息默認(rèn)是持久化的
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
return message;
}
});
}
}
測(cè)試
@Test
void test5() throws IOException {
for (int i = 1; i <=5;i++){
OrderingOk orderingOk = OrderingOk.builder().id(i).name("張 " + i).build();
delayProvider.send(orderingOk);
System.out.println("發(fā)送成功:"+i);
}
System.in.read();
}到此這篇關(guān)于rabbitmq延遲隊(duì)列的使用的文章就介紹到這了,更多相關(guān)rabbitmq延遲隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
- RabbitMQ 實(shí)現(xiàn)延遲隊(duì)列的兩種方式詳解
- RabbitMQ死信機(jī)制實(shí)現(xiàn)延遲隊(duì)列的實(shí)戰(zhàn)
- 手把手帶你掌握SpringBoot RabbitMQ延遲隊(duì)列
- 如何通過Python實(shí)現(xiàn)RabbitMQ延遲隊(duì)列
- RabbitMQ延遲隊(duì)列及消息延遲推送實(shí)現(xiàn)詳解
- Rabbitmq延遲隊(duì)列實(shí)現(xiàn)定時(shí)任務(wù)的方法
- Spring Boot與RabbitMQ結(jié)合實(shí)現(xiàn)延遲隊(duì)列的示例
- C#實(shí)現(xiàn)rabbitmq 延遲隊(duì)列功能實(shí)例代碼
相關(guān)文章
SpringBoot整合Dubbo+Zookeeper實(shí)現(xiàn)RPC調(diào)用
這篇文章主要給大家介紹了Spring Boot整合Dubbo+Zookeeper實(shí)現(xiàn)RPC調(diào)用的步驟詳解，文中有詳細(xì)的代碼示例，對(duì)我們的學(xué)習(xí)或工作有一定的幫助,需要的朋友可以參考下2023-07-07
dubbo服務(wù)無法注冊(cè)到zookeeper的問題
Dubbo+Zookeeper項(xiàng)目服務(wù)注冊(cè)失敗,因模塊創(chuàng)建時(shí)誤選Web項(xiàng)目導(dǎo)致main下文件夾名為data,服務(wù)未被識(shí)別,更改為java后,服務(wù)正常注冊(cè),Zookeeper心跳及接口調(diào)用問題解決2025-07-07
springmvc下實(shí)現(xiàn)登錄驗(yàn)證碼功能示例
本篇文章主要介紹了springmvc下實(shí)現(xiàn)登錄驗(yàn)證碼功能示例，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2017-02-02
java簡(jiǎn)單實(shí)現(xiàn)桌球滾動(dòng)效果
這篇文章主要為大家詳細(xì)介紹了java簡(jiǎn)單實(shí)現(xiàn)桌球滾動(dòng)效果，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-10-10
使用Java和SNMP4J實(shí)現(xiàn)SNMP操作完整代碼
這篇文章主要介紹了如何使用Java和SNMP4J庫進(jìn)行SNMP操作,包括初始化SNMP、創(chuàng)建目標(biāo)、創(chuàng)建PDU、發(fā)送SNMP請(qǐng)求和處理響應(yīng)等內(nèi)容,通過編寫SnmpUtil類,展示了完整的SNMP操作流程,需要的朋友可以參考下2024-12-12
Spring Cloud與分布式系統(tǒng)簡(jiǎn)析
這篇文章主要介紹了Spring Cloud與分布式系統(tǒng)的相關(guān)內(nèi)容，具有一定參考價(jià)值，需要的朋友可以了解下。2017-09-09
這篇文章主要介紹了IDEA安裝阿里代碼規(guī)范插件的步驟，本文通過圖文并茂的形式給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-12-12
這篇文章主要介紹了使用迭代器Iterator遍歷Collection問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-11-11
kafka?消息隊(duì)列中點(diǎn)對(duì)點(diǎn)與發(fā)布訂閱的區(qū)別說明
這篇文章主要介紹了kafka?消息隊(duì)列中點(diǎn)對(duì)點(diǎn)與發(fā)布訂閱的區(qū)別說明，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-05-05
Mybatis Properties 配置優(yōu)先級(jí)詳解
這篇文章主要介紹了Mybatis Properties 配置優(yōu)先級(jí)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-07-07