- 延迟队列
- 概念
- 使用场景
- RabbitMQ 中的 TTL
- 队列设置 TTL
- 消息设置 TTL
- 两者的区别
- 整合 springboot
- pom文件
- 配置文件
- 添加Swagger配置类
- 队列 TTL
- 代码架构
- 代码实现
- 延时队列优化
- 代码架构图
- 实现
- Rabbitmq 插件实现延迟队列
- 安装延时队列插件
- 代码架构图
- 总结
- 发布确认高级
- 介绍
- 代码架构
- 代码
- 回退消息
- Mandatory 参数
- 配置文件中开启回退消息功能
- 设置回退接口
- 发布测试
- 小总结
- 备份交换机
- 代码实现
- 配置类
- 注意:mandatory 参数与备份交换机可以一起使用的时候，如果两者同时开启，消息究竟何去何从？谁优先级高，经过上面结果显示答案是备份交换机优先级高
延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。
使用场景1.订单在十分钟之内未支付则自动取消
2.新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
3.用户注册成功后，如果三天内没有登陆则进行短信提醒。
4.用户发起退款，如果三天内没有得到处理则通知相关运营人员。
5.预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议
这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，
如：
发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；
看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？
如果数据量比较少，确实可以这样做，
比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。
但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单
的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。
RabbitMQ 中的 TTL
TTL 是什么呢？TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间。
单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的 TTL 和消息的TTL，那么较小的那个值将会被使用，有两种方式设置 TTL。
队列设置 TTL第一种是在创建队列的时候设置队列的“x-message-ttl”属性
消息设置 TTL另一种方式便是针对每条消息设置 TTL
两者的区别如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)，
而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；
另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，
如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。
整合 springboot pom文件
配置文件org.springframework.boot spring-boot-starterorg.springframework.boot spring-boot-starter-testtest org.springframework.boot spring-boot-starter-amqporg.springframework.boot spring-boot-starter-webcom.alibaba fastjson1.2.76 org.projectlombok lombokio.springfox springfox-swagger23.0.0 io.springfox springfox-swagger-ui3.0.0 org.springframework.amqp spring-rabbit-testtest
spring.rabbitmq.host=192.168.112.128 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123添加Swagger配置类
@Configuration
@EnableSwagger2
public class SwaggerConfig {
@Bean
public Docket webApiConfig(){
return new Docket(documentationType.SWAGGER_2)
.groupName("webApi")
.apiInfo(webApiInfo())
.select()
.build();
}
private ApiInfo webApiInfo(){
return new ApiInfoBuilder()
.title("rabbitmq接口文档")
.description("本文档描述了rabbitmq微服务接口定义")
.version("1.0")
.contact(new Contact("dhy","www.baidu.com","123456@qq.com"))
.build();
}
}
队列 TTL 代码架构
创建两个队列QA和QB，两者队列TTL分别设置为10S和40S，然后在创建一个交换机X和死信交换机Y，它们的类型都是direct，创建一个死信队列QD，它们的绑定关系如下：
代码实现
配置类代码
/*
* TTL队列 配置文件类代码
*
* */
@Configuration
public class TtlQueueConfig {
//普通交换机的名称
public static final String X_EXCHANGE = "X";
//死信交换机的名称
public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
//普通队列的名称
public static final String QUEUE_A = "QA";
public static final String QUEUE_B = "QB";
//死信队列的名称
public static final String DEAD_LATTER_QUEUE = "QD";
//声明xExchange---普通交换机
@Bean("xExchange")
public DirectExchange xExchange()
{
return new DirectExchange(X_EXCHANGE);
}
//声明yExchange---死信交换机
@Bean("yExchange")
public DirectExchange yExchange()
{
return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
}
//声明队列---创建并设置普通队列相关属性
@Bean("queueA")
public Queue queueA(){
Map arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信Routing-key
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms ----该队列里面消息过期时间为10s
arguments.put("x-message-ttl",10000);
//队列持久化
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
}
//声明普通队列 TTL为40s
@Bean("queueB")
public Queue queueB(){
Map arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信Routing-key
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms
arguments.put("x-message-ttl",40000);
return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}
//死信队列
@Bean("queueD")
public Queue queueD(){
return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
}
//绑定
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
@Qualifier("xExchange") DirectExchange xExchange){
//队列A和X交换机绑定
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
//绑定
@Bean
public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
@Qualifier("xExchange") DirectExchange xExchange){
//队列B和x交换机绑定
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
//绑定
@Bean
public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
@Qualifier("yExchange") DirectExchange yExchange){
//死信队列D和死信交换机绑定
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
消费者代码：定义rabbitMq的消费者可以相当方便，只需要在消息处理类或者类方法加上@RabbitListener注解，指定队列名称即可。
/*
* 队列TTL 消费者
* */
@Slf4j
@Component
public class DeadLetterQueueConsumer {
//接收消息
@RabbitListener(queues = "QD")
public void receiveD(Message message, Channel channel) throws Exception {
String msg = new String(message.getBody());
log.info("当前时间：{}，收到死信队列的消息：{}",new Date().toString(),msg);
}
}
生产者代码
/*
* 发送延迟消息
* */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
//开始发消息
@GetMapping("/sendMsg/{message}")
public void sendMsg(@PathVariable String message){
log.info("当前时间：{}，发送一条信息给两个TTL队列：{}",new Date().toString(),message);
//String exchange, String routingKey, Object object
rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);
}
}
效果
第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 40S 之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了
不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只有 10S 和 40S两个时间选项，如果需要一个小时后处理，那么就需要增加 TTL 为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？
延时队列优化 代码架构图
在这里新增了一个队列 QC,绑定关系如下,该队列不设置 TTL 时间
因为我们知道，ttl可以在生产者这边设置，也可以在消费者这边设置，如果是在消费者这边设置，这针对的是当前队列里面的所有消息，如果是在生产者这边设置，针对的是发出去的每一条消息
因此这里我们新增一个队列，在消费者这边不设置ttl,而是在生产者发出消息的时候，动态调整ttl时间
实现
配置文件类，新增队列QC
/*
* TTL队列 配置文件类代码
*
* */
@Configuration
public class TtlQueueConfig {
//普通交换机的名称
public static final String X_EXCHANGE = "X";
//死信交换机的名称
public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
//普通队列的名称
public static final String QUEUE_A = "QA";
public static final String QUEUE_B = "QB";
//优化队列QC
public static final String QUEUE_C = "QC";
//死信队列的名称
public static final String DEAD_LATTER_QUEUE = "QD";
//声明xExchange---普通交换机
@Bean("xExchange")
public DirectExchange xExchange()
{
return new DirectExchange(X_EXCHANGE);
}
//声明yExchange---死信交换机
@Bean("yExchange")
public DirectExchange yExchange()
{
return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
}
//声明队列---创建并设置普通队列相关属性
@Bean("queueA")
public Queue queueA(){
Map arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信Routing-key
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms ----该队列里面消息过期时间为10s
arguments.put("x-message-ttl",10000);
//队列持久化
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
}
//声明普通队列 TTL为40s
@Bean("queueB")
public Queue queueB(){
Map arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信Routing-key
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms
arguments.put("x-message-ttl",40000);
return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}
//声明QC队列
@Bean("queueC")
public Queue queueC(){
Map arguments = new HashMap<>();
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信RoutingKey
arguments.put("x-dead-letter-routing-key","YD");
return QueueBuilder.durable().withArguments(arguments).build();
}
//死信队列
@Bean("queueD")
public Queue queueD(){
return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
}
//绑定
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
@Qualifier("xExchange") DirectExchange xExchange){
//队列A和X交换机绑定
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
//绑定
@Bean
public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
@Qualifier("xExchange") DirectExchange xExchange){
//队列B和x交换机绑定
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
//绑定
@Bean
public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueC).to(xExchange).with("XC");
}
//绑定
@Bean
public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
@Qualifier("yExchange") DirectExchange yExchange){
//死信队列D和死信交换机绑定
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
生产者
/*
* 发送延迟消息
* */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
//开始发消息
@GetMapping("/sendMsg/{message}")
public void sendMsg(@PathVariable String message){
log.info("当前时间：{}，发送一条信息给两个TTL队列：{}",new Date().toString(),message);
//String exchange, String routingKey, Object object
rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);
}
//开始发消息
@GetMapping("sendExpirationMsg/{message}/{ttlTime}")
public void sendMsg(@PathVariable String message,@PathVariable String ttlTime){
log.info("当前时间：{}，发送一条时长{}毫秒TTL信息给队列QC：{}",
new Date().toString(),ttlTime,message);
//通过x交换机加上XC路由key，将消息放到指定队列中，并同时设置当前发送消息的ttl
rabbitTemplate.convertAndSend("X","XC",message,msg->{
//发送消息的时候 延迟时长
msg.getMessageProperties().setExpiration(ttlTime);
return msg;
});
}
}
消费者代码无需改变，死信队列代码无需改变
测试
http://localhost:8080/ttl/sendExpirationMsg/你好 1/20000
http://localhost:8080/ttl/sendExpirationMsg/你好 2/2000
看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。
Rabbitmq 插件实现延迟队列
上文中提到的问题，确实是一个问题，如果不能实现在消息粒度上的 TTL，并使其在设置的 TTL 时间及时死亡，就无法设计成一个通用的延时队列。那如何解决呢，接下来我们就去解决该问题。
安装延时队列插件在官网上下载 https://www.rabbitmq.com/community-plugins.html，下载rabbitmq_delayed_message_exchange 插件，然后解压放置到 RabbitMQ 的插件目录。
进入 RabbitMQ 的安装目录下的 plgins 目录，执行下面命令让该插件生效，然后重启 RabbitMQ
将插件拷贝到下面这个目录下面 /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
安装延迟插件 rabbitmq-plugins enable rabbitmq_delayed_message_exchange
重启RabbitMQ systemctl restart rabbitmq-server
以前:
现在:
代码架构图
在这里新增了一个队列 delayed.queue,一个自定义交换机 delayed.exchange，绑定关系如下:
配置文件类代码
@Configuration
public class DelayedQueueConfig {
//队列
public static final String DELAYED_QUEUE_NAME = "delayed.queue";
//交换机
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
//routingKey
public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
//声明队列
@Bean
public Queue delayedQueue(){
return new Queue(DELAYED_QUEUE_NAME);
};
//自定义交换机 我们在这里定义的是一个延迟交换机
@Bean
public CustomExchange delayedExchange(){
Map arguments = new HashMap<>();
//自定义交换机的类型
arguments.put("x-delayed-type","direct");
/*
* 1.交换机的名称
* 2.交换机的类型
* 3.是否需要持久化
* 4.是否需要自动删除
* 5.其他的参数
* */
return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",
true,false,arguments);
}
//绑定
@Bean
public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
@Qualifier("delayedExchange") CustomExchange delayedExchange){
//noargs()构建
return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
}
}
消费者
// 消费者代码 基于插件的延迟消息
@Slf4j
@Component
public class DelayQueueConsumer {
//监听消息---监听哪一个队列
@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
public void recieveDelayQueue(Message message){
String msg = new String(message.getBody());
log.info("当前时间：{}，收到延迟队列的消息：{}",new Date().toString(),msg);
}
}
生产者
/*
* 发送延迟消息
* */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
//开始发消息 基于插件的 消息 及 延迟的时间
@GetMapping("/sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime){
log.info("当前时间：{}，发送一条时长{}毫秒信息给延迟队列delayed.queue：{}",
new Date().toString(),delayTime,message);
rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME
,DelayedQueueConfig.DELAYED_ROUTING_KEY,message,msg -> {
// 发送消息的时候 延迟时长 单位ms
msg.getMessageProperties().setDelay(delayTime);
return msg;
});
}
}
发起请求：
http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000
第二个消息被先消费掉了，符合预期
总结
延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz或者利用 kafka 的时间轮，这些方式各有特点,看需要适用的场景
发布确认高级 介绍
在生产环境中由于一些不明原因，导致 rabbitmq重启，在RabbitMQ重启期间生产者消息投递失败,导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行RabbitMQ的消息可靠投递呢?特别是在这样比较极端的情况，RabbitMQ集群不可用的时候，无法投递的消息该如何处理呢：
代码架构
代码
配置类，声明交换机和队列，并进行绑定
// 配置类：发布确认（高级）
@Configuration
public class /confirm/iConfig {
//交换机
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i_exchange";
//队列
public static final String /confirm/i_QUEUE_NAME = "/confirm/i_queue";
//RoutingKey
public static final String /confirm/i_routing_key = "key1";
//声明交换机
@Bean//默认bean的名字是类名首字母小写
public DirectExchange /confirm/iExchange()
{
return new DirectExchange(/confirm/i_EXCHANGE_NAME);
}
//声明队列
@Bean//默认bean的名字是类名首字母小写
public Queue /confirm/iQueue()
{
return QueueBuilder.durable(/confirm/i_QUEUE_NAME).build();
}
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("/confirm/iQueue") Queue /confirm/iQueue,
@Qualifier("/confirm/iExchange")DirectExchange /confirm/iExchange)
{
return BindingBuilder.bind(/confirm/iQueue).to(/confirm/iExchange).with(/confirm/i_routing_key);
}
}
消息生产者
// 开始发消息 测试确认
@RestController
@Slf4j
@RequestMapping("//confirm/i")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
//发消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME
,/confirm/iConfig./confirm/i_routing_key
,message);
log.info("发送消息内容：{}",message);
}
}
消费者
// 接收消息
@Slf4j
@Component
public class Consumer {
//监听的队列
@RabbitListener(queues = /confirm/iConfig./confirm/i_QUEUE_NAME)
public void receive/confirm/iMessage(Message message){
String msg = new String(message.getBody());
log.info("接受到的队列/confirm/i.queue消息：{}",msg);
}
}
回调接口
//回调接口
@Component
@Slf4j
//实现回调接口
public class MyCallBack implements RabbitTemplate./confirm/iCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
//Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
@PostConstruct
public void init(){
//将当前实现类，注入到rabbitTemplate中的接口上
//不然到时候rabbitTemplate调用接口时，找不到我们的实现类
rabbitTemplate.set/confirm/iCallback(this);
}
/*
* 交换机确认回调方法,发消息后，交换机接收到了就回调
* 1.1 correlationData：保存回调消息的ID及相关信息
* 1.2 b:交换机收到消息，为ack=true
* 1.3 s:失败原因，成功为null
*
* 发消息，交换机接受失败，也回调
* 2.1 correlationData：保存回调消息的ID及相关信息
* 2.2 b:交换机没收到消息，为ack=false
* 2.3 s:失败的原因
*
* */
@Override
public void /confirm/i(CorrelationData correlationData, boolean b, String s) {
String id = correlationData!=null ? correlationData.getId():"";
if (b){
log.info("交换机已经收到ID为：{}的信息",id);
}else {
log.info("交换机还未收到ID为：{}的消息，由于原因：{}",id,s);
}
}
}
增加配置文件相关属性设置和消息发送方
@RestController
@Slf4j
@RequestMapping("//confirm/i")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
//发消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message)
{
//消息唯一标识
CorrelationData correlationData = new CorrelationData("1");
//交换机的名称，交换机的路由，消息和消息的唯一标识
rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME
,/confirm/iConfig./confirm/i_routing_key
,message,correlationData);
log.info("发送消息内容：{}",message);
}
}
在配置文件当中需要添加
spring.rabbitmq.publisher-/confirm/i-type=correlated
⚫ NONE 禁用发布确认模式，是默认值
⚫ CORRELATED 发布消息成功到交换器后会触发回调方法
⚫ SIMPLE
经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法
其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker
最后一个SIMPLE就是单个确认模式，即发布一条消息，确认一条消息
spring.rabbitmq.host=192.168.112.128 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123 spring.rabbitmq.publisher-/confirm/i-type=correlated
结果
可以看到，发送了两条消息，第一条消息的 RoutingKey 为 “key1”，第二条消息的 RoutingKey 为"key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。
回退消息
当前问题: 当交换机收到消息后，会调用回调函数，同样当交换机没收到消息时，也会调用回调函数，但是如果交换机收到了消息，队列没有收到消息，我们无法得知
Mandatory 参数在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。
那么如何让无法被路由的消息帮我想办法处理一下？
最起码通知我一声，我好自己处理啊。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。
配置文件中开启回退消息功能
spring.rabbitmq.publisher-returns=true设置回退接口
//回调接口
@Component
@Slf4j
//实现回调接口
public class MyCallBack implements RabbitTemplate./confirm/iCallback,RabbitTemplate.ReturnsCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
//Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
@PostConstruct
public void init(){
//将当前实现类，注入到rabbitTemplate中的接口上
//不然到时候rabbitTemplate调用接口时，找不到我们的实现类
rabbitTemplate.set/confirm/iCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/*
* 交换机确认回调方法,发消息后，交换机接收到了就回调
* 1.1 correlationData：保存回调消息的ID及相关信息
* 1.2 b:交换机收到消息，为ack=true
* 1.3 s:失败原因，成功为null
*
* 发消息，交换机接受失败，也回调
* 2.1 correlationData：保存回调消息的ID及相关信息
* 2.2 b:交换机没收到消息，为ack=false
* 2.3 s:失败的原因
*
* */
@Override
public void /confirm/i(CorrelationData correlationData, boolean b, String s) {
String id = correlationData!=null ? correlationData.getId():"";
if (b){
log.info("交换机已经收到ID为：{}的信息",id);
}else {
log.info("交换机还未收到ID为：{}的消息，由于原因：{}",id,s);
}
}
//可以在当消息传递过程中不可达目的的时将消息返回给生产者
//只有不可达目的地的时候才可回退
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
log.error("消息{}，被交换机{}退回，退回的原因：{},路由Key：{}",
new String(returnedMessage.getMessage().getBody())
, returnedMessage.getExchange()
, returnedMessage.getReplyText()
, returnedMessage.getRoutingKey());
}
}
发布测试
@RestController
@Slf4j
@RequestMapping("//confirm/i")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
//发消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
//交换机可达，队列可达
CorrelationData correlationData = new CorrelationData("1");
rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME
,/confirm/iConfig./confirm/i_routing_key
,message+"key1",correlationData);
log.info("发送消息内容：{}",message+"key1");
//下面这个交换机可达，队列不存在，不可达
CorrelationData correlationData2 = new CorrelationData("2");
rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME
,/confirm/iConfig./confirm/i_routing_key+"2"
,message+"key12",correlationData2);
log.info("发送消息内容：{}",message+"key12");
}
}
小总结
生产者将信道设置成/confirm/i模式，一旦信道进入/confirm/i模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）,这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。
对于/confirm/iCallback来说： 如果消息没有到exchange,则/confirm/i回调,ack=false 如果消息到达exchange,则/confirm/i回调,ack=true 对于ReturnCallback来说： exchange到queue成功,则不回调return exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了) 比如路由不到队列时触发回调
可以使用rabbitTemplate.setMandatory(true);替代配置文件中的spring.rabbitmq.publisher-returns=true
//Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
@PostConstruct
public void init(){
//将当前实现类，注入到rabbitTemplate中的接口上
//不然到时候rabbitTemplate调用接口时，找不到我们的实现类
rabbitTemplate.set/confirm/iCallback(this);
rabbitTemplate.setReturnsCallback(this);
/**
* true：
* 交换机无法将消息进行路由时，会将该消息返回给生产者
* false：
* 如果发现消息无法进行路由，则直接丢弃
*/
rabbitTemplate.setMandatory(true);
}
RabbitMQ自学之路（七）—— RabbitMQ消息发送确认与消息接收确认机制
RabbitMQ消息确认机制之/confirm/i模式总结
备份交换机
有了mandatory 参数和回退消息，我们获得了对无法投递消息的感知能力，有机会在生产者的消息无法被投递时发现并处理。
但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。
而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。
而且设置mandatory参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。
如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢?
前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。
在RabbitMQ.中，有一种备份交换机的机制存在，可以很好的应对这个问题。
什么是备份交换机呢?
备份交换机可以理解为 RabbitMQ中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为Fanout，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。
当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
代码实现 配置类
// 配置类：发布确认（高级）
@Configuration
public class /confirm/iConfig {
//交换机
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i_exchange";
//队列
public static final String /confirm/i_QUEUE_NAME = "/confirm/i_queue";
//RoutingKey
public static final String /confirm/i_routing_key = "key1";
//备份交换机
public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
//备份队列
public static final String BACKUP_QUEUE_NAME = "backup_queue";
//报警队列
public static final String WARNING_QUEUE_NAME = "warning_queue";
//声明交换机
@Bean//默认bean的名字是类名首字母小写
public DirectExchange /confirm/iExchange()
{
return ExchangeBuilder.directExchange(/confirm/i_EXCHANGE_NAME).durable(true)
//指定备份交换机
.withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
}
//声明队列
@Bean//默认bean的名字是类名首字母小写
public Queue /confirm/iQueue()
{
return QueueBuilder.durable(/confirm/i_QUEUE_NAME).build();
}
//备份交换机---一般是扇出类型
@Bean
public FanoutExchange backupExchange(){
return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
//备份队列
@Bean
public Queue backupQueue(){
return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
}
//报警队列
@Bean
public Queue warningQueue(){
return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
}
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("/confirm/iQueue") Queue /confirm/iQueue,
@Qualifier("/confirm/iExchange")DirectExchange /confirm/iExchange)
{
return BindingBuilder.bind(/confirm/iQueue).to(/confirm/iExchange).with(/confirm/i_routing_key);
}
@Bean
public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange){
return BindingBuilder.bind(backupQueue).to(backupExchange);
}
@Bean
public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange){
return BindingBuilder.bind(backupQueue).to(backupExchange);
}
}
消费者
// 报警消费者
@Component
@Slf4j
public class WarningConsumer {
//接受报警消息
@RabbitListener(queues = /confirm/iConfig.WARNING_QUEUE_NAME)
public void receiveWarningMsg(Message message){
String msg = new String(message.getBody());
log.error("报警发现不可路由消息：{}",msg);
}
}
重新启动项目的时候需要把原来的 /confirm/i.exchange 删除因为我们修改了其绑定属性，不然报以下错:
结果:
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）