本站精选了一篇相关的编程文章，网友李鸿畴根据主题投稿了本篇教程内容，涉及到rabbitMq死信队列、rabbitMq延时消息、rabbitMq延时队列、rabbitMq的死信队列实现延时消息相关内容，已被585网友关注，如果对知识点想更进一步了解可以在下方电子资料中获取。
rabbitMq的死信队列实现延时消息
前言
使用mq自带的死信去实现延时消息要注意一个坑点，就是mq只会检测队首的消息的过期时间，假设先放入队列10s过期消息，再放入2s过期。
mq会检测头部10s是否过期，10s不过期的情况下，2s就算过去也不会跑到死信.
mq基本的消息模型
mq死信队列的消息模型
简单的说就是先弄一个正常队列，然后不要设置消费者，接着给这个正常队列绑定一个死信队列，这个死信队列设置方式和正常队列没啥区别。
然后监听这个死信队列的消费.
一般死信队列由三大核心组件组成：死信交换机+死信路由+TTL（消息过期时间）
通常死信队列由“面向生产者的基本交换机+基本路由”绑定而成，所以生产者首先是将消息发送至“基本交换机+基本路由”所绑定而成的消息模型中，即间接性地进入到死信队列中，当过了TTL，消息将“挂掉”，从而进入下一个中转站，即“面下那个消费者的死信交换机+死信路由”所绑定而成的消息模型中.
maven依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
配置普通队列和死信队列
yml文件
spring:
application:
name: ttl-queue
rabbitmq:
host: 110.40.181.73
port: 35672
username: root
password: 10086
virtual-host: /fchan
connection-timeout: 15000
# 发送确认
#publisher-confirms: true
# 路由失败回调
publisher-returns: true
template:
# 必须设置成true 消息路由失败通知监听者，而不是将消息丢弃
mandatory: true
listener:
simple:
# 每次从RabbitMQ获取的消息数量
prefetch: 1
default-requeue-rejected: false
# 每个队列启动的消费者数量
concurrency: 1
# 每个队列最大的消费者数量
max-concurrency: 1
# 签收模式为手动签收-那么需要在代码中手动ACK
acknowledge-mode: manual
package com.fchan.mq.mqDelay.dlx;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MyRabConfig {
//定义普通任务队列（其实就是一个普通队列，只是没有消费者）
@Bean("delayQue")
public Queue delayQue(){
//普通队列绑定死信交换机及路由key
//delayQueue延时队列
return QueueBuilder.durable("delayQueue")
//指定这个延时队列下的死信交换机
//如果消息过时则会被投递到当前对应的my-dlx-exchange
.withArgument("x-dead-letter-exchange", "my-dlx-exchange")
//死信交换机绑定的路由key
//如果消息过时,my-dlx-exchange会根据routing-key-delay投递消息到对应的队列
//mq发送消息的时候一般指定的是exchange交换机+这个routingKey来找到队列，这个应该是出于有时候需要让mq的交换机将一个消息发送到多个队列所采用的方式
.withArgument("x-dead-letter-routing-key", "routing-key-delay")
.build();
}
//定义普通队列的交换机
@Bean("delayExchange")
public Exchange delayExchange(){
return ExchangeBuilder.directExchange("delayExchange").build();
}
//绑定普通队列的交换机
@Bean("delayBinding")
public Binding delayBinding(@Qualifier("delayExchange") Exchange exchange, @Qualifier("delayQue") Queue queue){
return BindingBuilder.bind(queue).to(exchange).with("delayBindingRoutingKey").noargs();
}
//定义死信队列
@Bean("dlxQueue")
public Queue dlxQueue(){
return QueueBuilder.durable("my-dlx-queue").build();
}
//定义死信交换机，这里的死信交换机要和上面普通队列所绑定的交换机名称一致
@Bean("dlxExchange")
public Exchange dlxExchange(){
return ExchangeBuilder.directExchange("my-dlx-exchange").build();
}
//绑定交换机与死信队列
@Bean("dlxBinding")
public Binding dlxBinding(@Qualifier("dlxExchange") Exchange exchange, @Qualifier("dlxQueue") Queue queue){
//这儿这个routingKey要和上面正常队列中所绑定的死信routingKey一致
return BindingBuilder.bind(queue).to(exchange).with("routing-key-delay").noargs();
}
}
死信队列消费者
package com.fchan.mq.mqDelay;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class MyRabbitConsume {
Logger log = LoggerFactory.getLogger(MyRabbitConsume.class);
@RabbitListener(queues = {"my-dlx-queue"})
public void receiveDeadMessage(Map<String,Object> map, Message message, Channel channel) throws Exception {
log.info("收到死信信息:{}",map);
log.info("然后进行一系列逻辑处理 Thanks♪(･ω･)ﾉ");
//手动确认消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
发送消息测试
/**
* 发送消息到延时队列 delayQueue
* 消息5秒后会被投递到死信队列
* @return
*/
@GetMapping("sendMsgToDelay")
public String sendMsgToDelay(String msg){
Map<String,Object> map = new HashMap<>();
map.put("msg", msg);
rabbitTemplate.convertAndSend("delayExchange","delayBindingRoutingKey", map, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//设置消息过期时间,5秒过期
message.getMessageProperties().setExpiration("5000");
return message;
}
});
return msg;
}
测试成功
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持码农之家。
