MQ的可靠性投递优化

call-back-service/src/main/java/cn/freemud/amp/service/SendMQConfirmCallback.java

+package cn.freemud.amp.service;
+
+import lombok.extern.log4j.Log4j2;
+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.beans.factory.annotation.Configurable;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * @author : xh.Z
+ * @email : fisherman0510@163.com
+ * @Date : 2022/8/19 14:40
+ * @description : 消息投递 确认机制
+ */
+@Configurable
+@Log4j2
+public class SendMQConfirmCallback {
+
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * 定制 RabbitTemplate
+     * <p>
+     * PostConstruct ：RabbitConfig对象创建完成之后执行这个方法
+     */
+    @PostConstruct
+    public void initRabbitTemplate() {
+
+        /**
+         * 设置消息到达服务器确认回调
+         */
+        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
+            /**
+             *
+             * @param correlationData 当前消息的唯一关联数据（消息的唯一ID）
+             * @param ack 交换机是否成功收到消息
+             * @param cause 失败信息
+             */
+            @Override
+            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+                log.info("消息ID：{}，消息是否到达Broker：{}，失败信息：{}", correlationData, ack, cause);
+            }
+        });
+
+        /**
+         * 消息没有抵达队列的回调
+         */
+        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
+            /**
+             * 消息没有投递给指定的队列，就会触发这个方法
+             *
+             * @param message 投递失败的消息详细信息
+             * @param replyCode 回复的状态码
+             * @param replyText 回复的文本内容
+             * @param exchange 目标交换机
+             * @param routingKey 使用的路由键
+             */
+            @Override
+            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
+                log.info("失败消息的信息：{}，状态码：{}，文本内容：{}，目标交换机：{}，使用的路由键：{}", message, replyCode, replyText, exchange, routingKey);
+            }
+        });
+    }
+
+}