湘潭网站建设 沟通磐石网络,北京国税局网站做票种核定,个人网站建设的国外文献综述,怎么上传网站RabbitMq的确认机制和延时通知
一、消息发送确认
在RabbitConfig中两个回调函数#xff0c;一个叫 ConfirmCallback #xff0c;一个叫 RetrunCallback#xff1b;
1、交换机确认#xff1a;ConfirmCallback方法
ConfirmCallback 是一个回调接口#xff0c;消息发送到…RabbitMq的确认机制和延时通知
一、消息发送确认
在RabbitConfig中两个回调函数一个叫 ConfirmCallback 一个叫 RetrunCallback
1、交换机确认ConfirmCallback方法
ConfirmCallback 是一个回调接口消息发送到 Broker 后触发回调确认消息是否到达 Broker 服务器也就是只确认是否正确到达 Exchange 中。
我们需要在生产者的配置中添加下面配置表示开启发布者确认。
spring.rabbitmq.publisher-confirm-typecorrelated # 新版本
spring.rabbitmq.publisher-confirmstrue # 老版本实现接口 ConfirmCallback 重写其confirm()方法方法内有三个参数correlationData、ack、cause。
correlationData对象内部只有一个 id 属性用来表示当前消息的唯一性。ack消息投递到broker 的状态true表示成功。cause表示投递失败的原因。
2、队列确认ReturnCallback方法
交换机接收到消息后可以判断当前的路径发送没有问题但是不能保证消息能够发送到路由队列的。而发送者是不知道这个消息有没有送达队列的因此我们需要在队列中进行消息确认。这就是回退消息。
实现接口ReturnCallback重写 returnedMessage() 方法方法有五个参数message消息体、replyCode响应code、replyText响应内容、exchange交换机、routingKey队列。
添加以下配置
spring.rabbitmq.publisher-returnstrue3、消息发送确认代码实现
在rabbitConfig中实现接口
package com.it520.bookkeeping.config;import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Author : huliupan* CreateTime : 2022-10-13* Description : RabbitMQ的config**/
Configuration
public class RabbitConfig {Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println(ConfirmCallback: 相关数据correlationData);System.out.println(ConfirmCallback: 确认情况ack);System.out.println(ConfirmCallback: 原因cause);}});rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {System.out.println(ReturnCallback: 消息returnedMessage.getMessage());System.out.println(ReturnCallback: 回应码returnedMessage.getReplyCode());System.out.println(ReturnCallback: 回应信息returnedMessage.getReplyText());System.out.println(ReturnCallback: 交换机returnedMessage.getExchange());System.out.println(ReturnCallback: 路由键returnedMessage.getRoutingKey());}});return rabbitTemplate;}}4、回调的触发情况
那么以上这两种回调函数都是在什么情况会触发呢
①消息推送到server但是在server里找不到交换机 ②消息推送到server找到交换机了但是没找到队列 ③消息推送到server交换机和队列啥都没找到 ④消息推送成功
那么我先写几个接口来分别测试和认证下以上4种情况消息确认触发回调函数的情况
消息推送到server但是在server里找不到交换机 写个测试接口把消息推送到名为‘non-existent-exchange’的交换机上这个交换机是没有创建没有配置的
GetMapping(/TestMessageAck)
public String TestMessageAck() {String messageId String.valueOf(UUID.randomUUID());String messageData message: non-existent-exchange test message ;String createTime LocalDateTime.now().format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss));MapString, Object map new HashMap();map.put(messageId, messageId);map.put(messageData, messageData);map.put(createTime, createTime);rabbitTemplate.convertAndSend(non-existent-exchange, TestDirectRouting, map);return ok;
}调用接口查看rabbitmq-provuder项目的控制台输出情况原因里面有说没有找到交换机’non-existent-exchange’
ConfirmCallback: 相关数据null
ConfirmCallback: 确认情况false
ConfirmCallback: 原因channel error; protocol method: #methodchannel.close(reply-code404, reply-textNOT_FOUND - no exchange non-existent-exchange in vhost /, class-id60, method-id40)
2022-10-14 16:15:58.721 ERROR 4868 --- [.98.153.34:5672] o.s.a.r.c.CachingConnectionFactory : Shutdown Signal: channel error; protocol method: #methodchannel.close(reply-code404, reply-textNOT_FOUND - no exchange non-existent-exchange in vhost /, class-id60, method-id40) 结论 ①这种情况触发的是 ConfirmCallback 回调函数。
消息推送到server找到交换机了但是没找到队列 这种情况就是需要新增一个交换机但是不给这个交换机绑定队列我来简单地在DirectRabitConfig里面新增一个直连交换机名叫‘lonelyDirectExchange’但没给它做任何绑定配置操作
Bean
DirectExchange lonelyDirectExchange() {return new DirectExchange(lonelyDirectExchange);
}然后写个测试接口把消息推送到名为‘lonelyDirectExchange’的交换机上这个交换机是没有任何队列配置的
GetMapping(/TestMessageAck2)
public String TestMessageAck2() {String messageId String.valueOf(UUID.randomUUID());String messageData message: lonelyDirectExchange test message ;String createTime LocalDateTime.now().format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss));MapString, Object map new HashMap();map.put(messageId, messageId);map.put(messageData, messageData);map.put(createTime, createTime);rabbitTemplate.convertAndSend(lonelyDirectExchange, TestDirectRouting, map);return ok;
}调用接口查看rabbitmq-provuder项目的控制台输出情况
ConfirmCallback: 相关数据null
ConfirmCallback: 确认情况true
ConfirmCallback: 原因null
ReturnCallback: 消息(Body:{createTime2022-10-14 16:17:47, messageIdc001cbbe-7792-465f-b6bd-cb6f3bdde27b, messageDatamessage: lonelyDirectExchange test message } MessageProperties [headers{}, contentTypeapplication/x-java-serialized-object, contentLength0, receivedDeliveryModePERSISTENT, priority0, deliveryTag0])
ReturnCallback: 回应码312
ReturnCallback: 回应信息NO_ROUTE
ReturnCallback: 交换机lonelyDirectExchange
ReturnCallback: 路由键TestDirectRouting可以看到这种情况两个函数都被调用了 这种情况下消息是推送成功到服务器了的所以ConfirmCallback对消息确认情况是true 而在RetrunCallback回调函数的打印参数里面可以看到消息是推送到了交换机成功了但是在路由分发给队列的时候找不到队列所以报了错误 NO_ROUTE 。 结论这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。
消息推送到sever交换机和队列啥都没找到 GetMapping(/TestMessageAck2)public String TestMessageAck2() {String messageId String.valueOf(UUID.randomUUID());String messageData message: lonelyDirectExchange test message ;String createTime LocalDateTime.now().format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss));MapString, Object map new HashMap();map.put(messageId, messageId);map.put(messageData, messageData);map.put(createTime, createTime);rabbitTemplate.convertAndSend(lonelyDirectExchange, TestDirectRouting, map);return ok;}返回结果和没有交换机的一样
022-10-14 16:19:11.882 ERROR 4868 --- [.98.153.34:5672] o.s.a.r.c.CachingConnectionFactory : Shutdown Signal: channel error; protocol method: #methodchannel.close(reply-code404, reply-textNOT_FOUND - no exchange non-existent-exchange in vhost /, class-id60, method-id40)
ConfirmCallback: 相关数据null
ConfirmCallback: 确认情况false
ConfirmCallback: 原因channel error; protocol method: #methodchannel.close(reply-code404, reply-textNOT_FOUND - no exchange non-existent-exchange in vhost /, class-id60, method-id40)消息推送成功
那么测试下按照正常调用之前消息推送的接口就行就调用下 /sendFanoutMessage接口可以看到控制台输出
ConfirmCallback: 相关数据null
ConfirmCallback: 确认情况true
ConfirmCallback: 原因null结论 这种情况触发的是 ConfirmCallback 回调函数。
二、消息消费确认机制
Springboot 的确认模式有三种配置如下
spring.rabbitmq.listener.simple.acknowledge-modemanualNONE 不确认 1、默认所有消息消费成功会不断的向消费者推送消息2、因为 rabbitmq 认为所有消息都被消费成功。所以队列中存在丢失消息风险。 AUTO自动确认 1、根据消息处理逻辑是否抛出异常自动发送 ack正常和nack异常给服务端如果消费者本身逻辑没有处理好这条数据就存在丢失消息的风险。2、使用自动确认模式时需要考虑的另一件事情就是消费者过载。 MANUAL手动确认 1、手动确认在业务失败后进行一些操作消费者调用 ack、nack、reject 几种方法进行确认如果消息未被 ACK 则发送到下一个消费者或重回队列。2、ack 用于肯定确认nack 用于 否定确认 reject 用于否定确认一次只能拒绝单条消息
1、自动确认
自动确认是指消费者在消费消息的时候当消费者收到消息后消息就会被 RabbitMQ 从队列中删除掉。这种模式认为 “发送即成功”。这是不安全的因为消费者可能在业务中并没有成功消费完就中断了
2、手动确认
手动确认又分为肯定确认和否定确认。
2.1 basicAck 方法肯定确认
basicAck 方法用于确认当前消息Channel 类中的 basicAck 方法定义如下
void basicAck(long deliveryTag, boolean multiple) throws IOException;参数说明
long deliveryTag唯一标识 ID当一个消费者向 RabbitMQ 注册后会建立起一个 Channel RabbitMQ 会用 basic.deliver 方法向消费者推送消息这个方法携带了一个 delivery tag 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID是一个单调递增的正整数delivery tag 的范围仅限于 Channel。
boolean multiple是否批处理当该参数为 true 时则可以一次性确认 delivery_tag 小于等于传入值的所有消息。
2.2 basicNack 方法否定确认
basicNack 方法用于否定当前消息。 由于 basicReject 方法一次只能拒绝一条消息如果想批量拒绝消息则可以使用 basicNack 方法。消费者客户端可以使用 channel.basicNack 方法来实现方法定义如下
void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;参数说明
long deliveryTag唯一标识 ID。
boolean multiple是否批处理当该参数为 true 时则可以一次性确认 delivery_tag 小于等于传入值的所有消息。
boolean requeue如果 requeue 参数设置为 true则 RabbitMQ 会重新将这条消息存入队列以便发送给下一个订阅的消费者 如果 requeue 参数设置为 false则 RabbitMQ 立即会还把消息从队列中移除而不会把它发送给新的消费者。
2.3 basicReject 方法否定确认
basicReject 方法用于明确拒绝当前的消息而不是确认。 RabbitMQ 在 2.0.0 版本开始引入 Basic.Reject 命令消费者客户端可以调用与其对应的 channel.basicReject 方法来告诉 RabbitMQ 拒绝这个消息。
Channel 类中的basicReject 方法定义如下
void basicReject(long deliveryTag, boolean requeue) throws IOException;参数说明
long deliveryTag唯一标识 ID。
boolean requeue上面已经解释。
利用之前的Fanout交换机的消息发送来测试消息确认
package com.it520.bookkeeping.receiver;import com.it520.bookkeeping.config.FanoutRabbitConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** Author : huliupan* CreateTime : 2022/10/13* Description :Fanout交换机的消费者**/
Component
public class FanoutReceiver {RabbitHandlerRabbitListener(queues FanoutRabbitConfig.FANOUT_QUEUE_A)public void processA(Map testMessage, Message message, Channel channel) throws IOException, ClassNotFoundException {String consumerQueueName message.getMessageProperties().getConsumerQueue();long deliveryTag message.getMessageProperties().getDeliveryTag();if (FanoutRabbitConfig.FANOUT_QUEUE_A.equals(consumerQueueName)) {/*** 确认消息参数说明* long deliveryTag唯一标识 ID。* boolean multiple是否批处理当该参数为 true 时* 则可以一次性确认 deliveryTag 小于等于传入值的所有消息。*/channel.basicAck(deliveryTag, true);System.out.println(fanout.a收到肯定确认了 deliveryTag);}}RabbitHandlerRabbitListener(queues FanoutRabbitConfig.FANOUT_QUEUE_B)public void processB(Map testMessage, Message message, Channel channel) throws IOException {String consumerQueueName message.getMessageProperties().getConsumerQueue();long deliveryTag message.getMessageProperties().getDeliveryTag();if (FanoutRabbitConfig.FANOUT_QUEUE_B.equals(consumerQueueName)) {/*** 否定消息参数说明* long deliveryTag唯一标识 ID。* boolean multiple是否批处理当该参数为 true 时* 则可以一次性确认 deliveryTag 小于等于传入值的所有消息。* boolean requeue如果 requeue 参数设置为 true* 则 RabbitMQ 会重新将这条消息存入队列以便发送给下一个订阅的消费者* 如果 requeue 参数设置为 false则 RabbitMQ 立即会还把消息从队列中移除* 而不会把它发送给新的消费者。*/channel.basicNack(deliveryTag, true, false);System.out.println(fanout.B收到否定确认了 deliveryTag 未重新放入队列 );}}RabbitHandlerRabbitListener(queues FanoutRabbitConfig.FANOUT_QUEUE_C)public void processC(Map testMessage, Message message, Channel channel) throws IOException, InterruptedException {String consumerQueueName message.getMessageProperties().getConsumerQueue();long deliveryTag message.getMessageProperties().getDeliveryTag();Thread.sleep(5000);if (FanoutRabbitConfig.FANOUT_QUEUE_C.equals(consumerQueueName)) {/*** 拒绝消息参数说明* long deliveryTag唯一标识 ID。* boolean requeue如果 requeue 参数设置为 true* 则 RabbitMQ 会重新将这条消息存入队列以便发送给下一个订阅的消费者* 如果 requeue 参数设置为 false则 RabbitMQ 立即会还把消息从队列中移除* 而不会把它发送给新的消费者。*/channel.basicReject(deliveryTag, true);System.out.println(fanout.C收到否定确认了 deliveryTag 重新放入队列 );}}/*** Fanout.c的QUEUE的第二个消费者避免无限循环* param testMessage* param message* param channel* throws IOException* throws InterruptedException*/RabbitHandlerRabbitListener(queues FanoutRabbitConfig.FANOUT_QUEUE_C)public void processC2(Map testMessage, Message message, Channel channel) throws IOException, InterruptedException {String consumerQueueName message.getMessageProperties().getConsumerQueue();long deliveryTag message.getMessageProperties().getDeliveryTag();Thread.sleep(5000);if (FanoutRabbitConfig.FANOUT_QUEUE_C.equals(consumerQueueName)) {channel.basicAck(deliveryTag, true);System.out.println(fanout.C2收到keng定确认了 deliveryTag );}}}返回的结果
fanout.a收到肯定确认了1
fanout.B收到否定确认了1未重新放入队列
fanout.C收到否定确认了1重新放入队列
fanout.C2收到keng定确认了1processC和processC2随机消费fanout.C队列多个消费者避免一个消费者重新放入队列造成的循环
三、实现延时队列
RabbitMQ本身是不支持延时队列的但是延时队列的使用还是很广泛的例如一个订单下单之后有30分钟的支付时间30分钟后要对订单的支付状态进行判断这时候就需要用到延时队列。
目前基于RabbitMq实现延时队列的方法有两种
1、死信队列
特性1、Time To Live(TTL)
RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl来控制消息的生存时间如果超时(两者同时设置以最先到期的时间为准)则消息变为dead letter(死信)
RabbitMQ针对队列中的消息过期时间有两种方法可以设置。 A: 通过队列属性设置队列中所有消息都有相同的过期时间。 B: 对消息进行单独设置每条消息TTL可以不同。 如果同时使用则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值就成为dead letter
特性2、Dead Letter ExchangesDLX
RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key可选两个参数如果队列内出现了dead letter则按照这两个参数重新路由转发到指定的队列。
x-dead-letter-exchange出现dead letter之后将dead letter重新发送到指定exchangex-dead-letter-routing-key出现dead letter之后将dead letter重新按照指定的routing-key发送
队列出现dead letter的情况有
消息或者队列的TTL过期队列达到最大长度消息被消费端拒绝basic.reject or basic.nack并且requeuefalse
综合上述两个特性设置了TTL规则之后当消息在一个队列中变成死信时利用DLX特性它能被重新转发到另一个Exchange或者Routing Key这时候消息就可以重新被消费了。
死信队列的配置
package com.it520.bookkeeping.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** Author : huliupan* CreateTime : 2022/10/13* Description :直连交换机的配置**/
Configuration
public class DirectRabbitConfig {public static final String DIRECT_EXCHANGE_NAME TestDirectExchange;public static final String DIRECT_QUEUE_NAME TestDirectQueue;public static final String DIRECT_QUEUE_NAME1 TestDirectQueue1;public static final String DIRECT_EXCHANGE_ROUTE_KEY TestDirectRouting;public static final String DIRECT_DELAY_EXCHANGE_NAME TestDirectDelayExchange;public static final String DIRECT_DELAY_QUEUE_NAME TestDirectDelayQueue;public static final String DIRECT_DELAY_QUEUE_NAME1 TestDirectDelayQueue1;public static final String DIRECT_DELAY_EXCHANGE_ROUTE_KEY TestDirectDelayRouting;//队列 起名TestDirectQueue//声明用于失效的队列的延时失效属性,并绑定到对应的死信交换机Beanpublic Queue TestDirectQueue() {MapString, Object args new HashMap(2);// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机args.put(x-dead-letter-exchange, DIRECT_DELAY_EXCHANGE_NAME);// x-dead-letter-routing-key 这里声明当前队列的死信路由keyargs.put(x-dead-letter-routing-key, DIRECT_DELAY_EXCHANGE_ROUTE_KEY);// x-message-ttl 声明队列的TTLargs.put(x-message-ttl, 1000 * 10);return new Queue(DIRECT_QUEUE_NAME, true, false, false, args);}//队列 起名TestDirectQueue1Beanpublic Queue TestDirectQueue1() {MapString, Object args new HashMap(2);// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机args.put(x-dead-letter-exchange, DIRECT_DELAY_EXCHANGE_NAME);// x-dead-letter-routing-key 这里声明当前队列的死信路由keyargs.put(x-dead-letter-routing-key, DIRECT_DELAY_EXCHANGE_ROUTE_KEY);// x-message-ttl 声明队列的TTLargs.put(x-message-ttl, 1000 * 5);return new Queue(DIRECT_QUEUE_NAME1, true, false, false, args);}//Direct交换机 起名TestDirectExchangeBeanDirectExchange TestDirectExchange() {// return new DirectExchange(TestDirectExchange,true,true);return new DirectExchange(DIRECT_EXCHANGE_NAME, true, false);}//声明延时交换机BeanDirectExchange createDelayExchange() {return new DirectExchange(DIRECT_DELAY_EXCHANGE_NAME, true, false);}//声明死信队列BeanQueue delayQueue() {return new Queue(DIRECT_DELAY_QUEUE_NAME, true);}//声明死信队列BeanQueue delayQueue1() {return new Queue(DIRECT_DELAY_QUEUE_NAME1, true);}//延时交换机和死信队列绑定BeanBinding bindingDelayQueue() {return BindingBuilder.bind(delayQueue()).to(createDelayExchange()).with(DIRECT_DELAY_EXCHANGE_ROUTE_KEY);}//延时交换机和死信队列绑定BeanBinding bindingDelayQueue1() {return BindingBuilder.bind(delayQueue1()).to(createDelayExchange()).with(DIRECT_DELAY_EXCHANGE_ROUTE_KEY);}//绑定 将队列和交换机绑定, 并设置用于匹配键TestDirectRoutingBeanBinding bindingDirect() {return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(DIRECT_EXCHANGE_ROUTE_KEY);}BeanBinding bindingDirect1() {return BindingBuilder.bind(TestDirectQueue1()).to(TestDirectExchange()).with(DIRECT_EXCHANGE_ROUTE_KEY);}BeanDirectExchange lonelyDirectExchange() {return new DirectExchange(lonelyDirectExchange);}}消息发送者 GetMapping(/sendDirectMessage)public String sendDirectMessage() {String messageData test message, hello!;//将消息携带绑定键值TestDirectRouting 发送到交换机TestDirectExchangerabbitTemplate.convertAndSend(DirectRabbitConfig.DIRECT_EXCHANGE_NAME, DirectRabbitConfig.DIRECT_EXCHANGE_ROUTE_KEY, messageData);return ok;}死信消息的接收者 RabbitHandlerRabbitListener(queues DirectRabbitConfig.DIRECT_DELAY_QUEUE_NAME)//监听的队列名称 TestDirectDelayQueuepublic void processDelay(String testMessage, Message message , Channel channel) throws IOException {long deliveryTag message.getMessageProperties().getDeliveryTag();System.out.println(死信消息properties message.getMessageProperties());System.out.println(DirectReceiver Delay消费者收到消息 : testMessage.toString());channel.basicAck(deliveryTag , false);}RabbitHandlerRabbitListener(queues DirectRabbitConfig.DIRECT_DELAY_QUEUE_NAME1)//监听的队列名称 TestDirectDelayQueue1public void processDelay1(String testMessage, Message message , Channel channel) throws IOException {long deliveryTag message.getMessageProperties().getDeliveryTag();System.out.println(死信消息properties message.getMessageProperties());System.out.println(DirectReceiver Delay1消费者收到消息 : testMessage.toString());channel.basicAck(deliveryTag , false);}
2、RabbitMq的插件
插件实现参考https://blog.csdn.net/u014308482/article/details/53036770
