springboot 集成 RabbitMQ confirm 确认模式和 return 回退模式以及Consumer Ack模式

说明: RabbitMQ消息的可靠投递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。

RabbitMQ 提供了两种方式用来控制消息的投递可靠性模式: confirm 确认模式, return 回退模式。

Consumer Ack: 表示消费端收到消息后的确认方式。

1. RabbitMQ confirm 确认模式和 return 回退模式

application.yml 配置文件开启confirm和return模式

spring:rabbitmq:host: 192.168.192.137username: qmhpassword: qmhvirtual-host: /qmhport: 5672# publisher-confirms 消息的可靠投递, confirm 确认模式 默认为falsepublisher-confirms: true# publisher-returns 消息的可靠投递, return 回退模式 默认为falsepublisher-returns: true

编写RabbitMQConfig配置类

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 RabbitMQConfig {public static final String DIRECT_EXCHANGE = "springboot_direct_exchange";public static final String DIRECT_QUEUE_CONFIRM = "springboot_direct_queue_confirm";  // 测试RabbitMQ消息的可靠投递// 1. 定义交换机@Bean("DirectExchange")public Exchange directExchange(){return ExchangeBuilder.topicExchange(DIRECT_EXCHANGE).durable(true).build();}// 2. 定义队列@Bean("DirectQueueConfirm")public Queue directQueueConfirm(){return QueueBuilder.durable(DIRECT_QUEUE_CONFIRM).build();}// 3. 绑定交换机与队列@Beanpublic Binding directBindQueueConfirmAndExchange(@Qualifier("DirectQueueConfirm") Queue directQueueConfirm,@Qualifier("DirectExchange") Exchange directExchange){return BindingBuilder.bind(directQueueConfirm).to(directExchange).with("confirm").noargs();}
}

编写测试类, 发送消息

import com.itheima.rabbitmqConfig.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@SpringBootTest
@RunWith(SpringRunner.class)
public class ProducerTest {@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 测试消息的可靠传递* confirm 确认模式* 当消息发送到exchange后回调confirm方法*/@Testpublic void directConfirmTest(){// 定义回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** @param correlationData   相关配置信息* @param ack       exchange交换机是否成功收到信息* @param cause     失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause)  {System.out.println("confirm running ...");if (ack){System.out.println("成功接收");}else {System.out.println("接收失败:" + cause);}}});// 发送消息rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE,"confirm","Test RabbitMQ confirm success!");}/*** 测试消息的可靠传递   return 回退模式* 当消息从exchange路由到queue失败后, 则会将消息退回给producer。并执行回调函数returnedMessage* 步骤:*  -- 开启回退模式   publisher-returns: true*  -- 设置ReturnCallBack回调方法*  -- 设置exchange交换机处理模式:*      1. 如果消息没有路由到Queue, 则丢弃消息(默认)*      2. 如果消息没有路由到Queue, 则返回消息给发送方ReturnCallBack.*/@Testpublic void directReturnTest(){// 设置交换机处理失败消息的模式// rabbitTemplate.setMandatory(true);// 设置ReturnCallBack回调方法rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** @param message       消息对象* @param replyCode     错误码* @param replyText     错误对象* @param exchange      交换机名称* @param routingKey    路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey){System.out.println("return running...");}});// 发送消息rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE,"confirm11","Test RabbitMQ return success!");}
}

2. RabbitMQ Consumer Ack 模式(下面案例是manual确认方式)

说明: Consumer Ack有三种确认方式

自动确认：acknowledge=“none”, (默认使用)

手动确认：acknowledge=“manual”, (建议使用)

根据异常情况确认：acknowledge=“auto”

application.yml 配置文件开启manual确认模式

spring:rabbitmq:host: 192.168.192.137username: qmhpassword: qmhvirtual-host: /qmhport: 5672# 开启manual确认模式listener:simple:acknowledge-mode: manual

编写消费者监听器, 主要实现ChannelAwareMessageListener类

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;/*** Consumer Ack 机制  ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。* 有三种确认方式：* -- 自动确认：acknowledge="none", 当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除* -- 手动确认：acknowledge="manual",需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。* -- 根据异常情况确认：acknowledge="auto"，*/
@Component
public class ListenerConsumerAck implements ChannelAwareMessageListener {@Override@RabbitListener(queues = "springboot_direct_queue_confirm")public void onMessage(Message message, Channel channel) throws Exception {Thread.sleep(1000);long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println(new String(message.getBody()));System.out.println("逻辑业务处理完成...");//int i = 3/0; //出错/*** 手动确认签收  basicAck(long deliveryTag, boolean multiple)* deliveryTag  标签* multiple     接收多条信息*/channel.basicAck(deliveryTag,true);}catch (Exception e){/*** 签收失败 basicNack(long deliveryTag, boolean multiple, boolean requeue)* deliveryTag  标签* multiple     接收多条信息* requeue      true,消息重回Queue,broker重新发送该消息给消费者*/channel.basicNack(deliveryTag,true,true);}}
}

springboot 集成 RabbitMQ confirm 确认模式和 return 回退模式以及Consumer Ack模式相关推荐

SpringBoot集成RabbitMQ消息队列搭建与ACK消息确认入门
1.Windows下安装RabbitMQ的步骤详解+图解(erlang+RabbitMQ) 2.SpringBoot集成RabbitMQ参考文章 1.RabbitMQ介绍 RabbitMQ是实现AMQ ...
Springboot集成rabbitMQ之mandatory和备份交换机
Springboot集成rabbitMQ之mandatory和备份交换机 mandatory 之前编写的消息队列代码中,通过重写ConfirmCallback中的confirm方法实现了消息送达的确认 ...
springboot 集成rabbitmq 实例
springboot 集成rabbitmq 实例个人在学习rabbitmq时发现网上很少有系统性介绍springboot和rabbitmq如何集成的,其他人总结的都片段化,所以结合个人调研过程,整理 ...
Springboot集成rabbitmq实现延时队列
Springboot集成rabbitmq实现延时队列什么是延时队列? 列举几个使用场景: 常见的种类有: 延时任务-实现方式: 详细信息:[https://www.cnblogs.com/JonaL ...
RabbitMQ——SpringBoot集成RabbitMQ
文章目录: 1.创建一个SpringBoot工程--消息发送者 1.创建一个SpringBoot工程--消息接收者 3.测试结果 3.1 direct 3.2 fanout 3.3 topic 3.4 ...
Springboot集成RabbitMQ一个完整案例
springboot 集成 RabbitMQ 非常简单,如果只是简单的使用配置非常少,springboot 提供了 spring-boot-starter-amqp 对消息各种支持. 1.配置pom文 ...
springboot集成rabbitmq商品秒杀业务实战（流量削峰）
消息队列如何实现流量削峰? 要对流量进行削峰,最容易想到的解决方案就是用消息队列来缓冲瞬时流量,把同步的直接调用转换成异步的间接推送,中间通过一个队列在一端承接瞬时的流量洪峰,在另一端平滑地将消息推送 ...
Net中使用 RabbitMq | Confirm确认消息
RabbitMQ 消息确认机制 - 消费者确认由于生产者和消费者不直接通信,生产者只负责把消息发送到队列,消费者只负责从队列获取消息(不管是push还是pull). 消息被"消费&quo ...
SpringBoot集成rabbitmq错误:org.springframework.amqp.AmqpConnectException: java.net.ConnectException的解决办法
在集成rabbitmq后,运行项目,报错日志: org.springframework.amqp.AmqpConnectException: java.net.ConnectException: Co ...

springboot 集成 RabbitMQ confirm 确认模式和 return 回退模式以及Consumer Ack模式

springboot 集成 RabbitMQ confirm 确认模式和 return 回退模式以及Consumer Ack模式

springboot 集成 RabbitMQ confirm 确认模式和 return 回退模式以及Consumer Ack模式相关推荐

最新文章

热门文章