ack

Author: yanpenglei

spring-boot-rabbitmq-ack/src/main/java/com/souyunku/example/springboot/rabbitmq/ack/producer/HelloSender.java

@@ -1,51 +1,70 @@
 package com.souyunku.example.springboot.rabbitmq.ack.producer;
 
+import com.souyunku.example.springboot.rabbitmq.ack.receiver.HelloReceiver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.util.Date;
 
 @Component
-public class HelloSender implements RabbitTemplate.ReturnCallback {
+public class HelloSender implements RabbitTemplate.ReturnCallback, RabbitTemplate.ConfirmCallback {
+
+    private Logger logger = LoggerFactory.getLogger(HelloSender.class);
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
     public void sendMessage(String context) {
-        System.out.println("HelloReceiver发送内容: " + context + "，发送时间:" + new Date());
-
 
         // 消息发送失败返回到队列中, application.properties 配置 spring.rabbitmq.publisher-returns=true
         rabbitTemplate.setMandatory(true);
 
         this.rabbitTemplate.setReturnCallback(this);
+        this.rabbitTemplate.setConfirmCallback(this);
 
         this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
             if (!ack) {
-                System.out.println("sendMessage 发送失败" + cause + correlationData.toString());
+                logger.info("HelloSender 发送失败：" + cause + correlationData.toString());
             } else {
-                System.out.println("sendMessage 发送成功 ");
+                logger.info("HelloSender 发送成功");
             }
         });
+
+        logger.info("HelloSender 发送的消息内容：{}", context);
+
         this.rabbitTemplate.convertAndSend("hello", context);
+
     }
 
 
     /**
      * 失败后返回消息回调
-     *
-     * @param message    消息返回的消息
-     * @param replyCode  回复代码
-     * @param replyText  回复文本
-     * @param exchange   交换交换
-     * @param routingKey 路由密钥
+     * <p>
+     * 当消息发送出去找不到对应路由队列时，将会把消息退回
+     * 如果有任何一个路由队列接收投递消息成功，则不会退回消息
      */
     @Override
     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
-        System.out.println("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
+        logger.info("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
     }
 
+    /**
+     * 实现ConfirmCallback
+     * <p>
+     * ACK=true仅仅标示消息已被Broker接收到，并不表示已成功投放至消息队列中
+     * ACK=false标示消息由于Broker处理错误，消息并未处理成功
+     */
+    @Override
+    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+        logger.info("消息id: " + correlationData + "确认" + (ack ? "成功:" : "失败"));
+    }
+
+
 
 }

spring-boot-rabbitmq-ack/src/main/java/com/souyunku/example/springboot/rabbitmq/ack/receiver/HelloReceiver.java

@@ -2,6 +2,8 @@
 
 
 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.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
@@ -15,20 +17,31 @@
 @RabbitListener(queues = "hello")
 public class HelloReceiver {
 
+    private Logger logger = LoggerFactory.getLogger(HelloReceiver.class);
+
     @RabbitHandler
-    public void process(String context, Message message, Channel channel) throws Exception {
-        System.out.println("HelloReceiver收到内容: " + context + "，收到时间:" + new Date());
+    public void process(String context, Message message, Channel channel) {
+        logger.info("HelloReceiver 监听到消息内容：{}", context);
         try {
+
             //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             //消息确认  因为我在属性配置文件里面开启了ACK确认 所以如果代码没有执行ACK确认 你在RabbitMQ的后台会看到消息会一直留在队列里面未消费掉 只要程序一启动开始接受该队列消息的时候 又会收到
 
-            System.out.println("HelloReceiver消息接收成功");
+            logger.info("HelloReceiver 消息接收成功");
         } catch (Exception e) {
             e.printStackTrace();
-            ///ack返回false，并重新回到队列，api里面解释得很清楚
-            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
-            System.out.println("消息接收失败");
+
+            logger.info("HelloReceiver 消息接收失败");
+            // ack返回false，并重新放回队列
+            try {
+                logger.info("HelloReceiver ack返回false，并重新放回队列");
+                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+            } catch (IOException e1) {
+                e1.printStackTrace();
+            }
+
+
         }
 
     }

spring-boot-rabbitmq-ack/src/main/resources/application.properties

@@ -6,6 +6,8 @@ spring.rabbitmq.host=10.4.98.15
 spring.rabbitmq.port=5672
 spring.rabbitmq.username=admin
 spring.rabbitmq.password=admin
+
+
 # 开启发送确认
 spring.rabbitmq.publisher-confirms=true
 # 开启发送失败退回