merge

boot-service/src/main/java/com/xwbing/rabbit/RabbitConfig.java

@@ -42,7 +42,7 @@ public class RabbitConfig {
         connectionFactory.setPassword(password);
         connectionFactory.setVirtualHost(virtualHost);
         connectionFactory.setConnectionTimeout(timeout);
-        // 如果要进行消息回调，则这里必须要设置为true
+        //如果要进行消息回调,则这里必须要设置为true
         connectionFactory.setPublisherConfirms(publisherConfirms);
         connectionFactory.setPublisherReturns(publisherReturns);
         return connectionFactory;
@@ -54,6 +54,8 @@ public class RabbitConfig {
     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
     @Bean
     public RabbitTemplate rabbitTemplate() {
-        return new RabbitTemplate(connectionFactory());
+        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
+        rabbitTemplate.setMandatory(true);
+        return rabbitTemplate;
     }
 }

boot-service/src/main/java/com/xwbing/rabbit/Receiver.java

@@ -5,6 +5,8 @@ import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
 
+import java.text.MessageFormat;
+
 import static com.xwbing.rabbit.RabbitConstant.HTTP_REQUEST_QUEUE;
 import static com.xwbing.rabbit.RabbitConstant.SERVER_INVOKE_QUEUE;
 
@@ -26,8 +28,8 @@ public class Receiver {
      */
     @RabbitListener(queues = SERVER_INVOKE_QUEUE)
     public String processServer(String msg) {
-        logger.info("接收到来自{}队列的消息:{}", SERVER_INVOKE_QUEUE, msg);
-        return msg.toUpperCase();
+        String response = MessageFormat.format("收到{0}队列的消息:{1}", SERVER_INVOKE_QUEUE, msg);
+        return response.toUpperCase();
     }
 
     /**
@@ -36,7 +38,8 @@ public class Receiver {
      * @param msg
      */
     @RabbitListener(queues = HTTP_REQUEST_QUEUE)
-    public void processHttp(String msg) {
-        logger.info("接收到来自{}队列的消息:{}", HTTP_REQUEST_QUEUE, msg);
+    public String processHttp(String msg) {
+        String response = MessageFormat.format("收到{0}队列的消息:{1}", HTTP_REQUEST_QUEUE, msg);
+        return response.toUpperCase();
     }
 }

boot-service/src/main/java/com/xwbing/rabbit/Sender.java

@@ -25,30 +25,31 @@ public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.Re
 
     @PostConstruct
     public void init() {
-        //用于实现消息发送到RabbitMQ交换器后接收ack回调。
+        //用于实现消息发送到RabbitMQ交换机后接收ack回调。
         rabbitTemplate.setConfirmCallback(this);
-        //用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
+        //用于实现消息发送到RabbitMQ交换机，但无相应队列与交换器绑定时的回调。
         rabbitTemplate.setReturnCallback(this);
     }
 
     /**
-     * 相应交换器接收后回调
+     * 相应交换机接收后异步回调
      */
     @Override
     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
         if (ack) {
-            logger.info("消息发送成功:{}", correlationData);
+            logger.info("交换机接收信息成功:{}", correlationData);
         } else {
-            logger.info("消息发送失败:{}", cause);
+            logger.info("交换机接收信息失败:{}", cause);
         }
     }
 
     /**
-     * 无交换器绑定回调
+     * 无相应队列与交换器绑定异步回调
      */
     @Override
     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
-        logger.info(message.getMessageProperties().getCorrelationIdString() + "发送失败");
+        String msg = new String(message.getBody());
+        logger.info("消息发送失败:{}", msg);
     }
 
     public void sendServerInvoke(String msg) {
@@ -70,8 +71,10 @@ public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.Re
         CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
         logger.info("开始发送消息:{}", msg.toLowerCase());
         //转换并发送消息,且等待消息者返回响应消息。
-        String response = rabbitTemplate.convertSendAndReceive(exchange, routingKey, msg, correlationId).toString();
+        Object response = rabbitTemplate.convertSendAndReceive(exchange, routingKey, msg, correlationId);
+        if (response != null) {
+            logger.info("消费者响应:{}", response.toString());
+        }
         logger.info("结束发送消息:{}", msg.toLowerCase());
-        logger.info("消费者响应:{}消息处理完成", response);
     }
 }

boot-web/src/main/java/com/xwbing/controller/rest/TestControl.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.xwbing.annotation.LogInfo;
 import com.xwbing.domain.entity.ExpressInfo;
 import com.xwbing.domain.entity.vo.ExpressInfoVo;
+import com.xwbing.rabbit.Sender;
 import com.xwbing.redis.RedisService;
 import com.xwbing.service.rest.CookieSessionService;
 import com.xwbing.service.rest.ExpressDeliveryService;
@@ -43,8 +44,15 @@ public class TestControl {
     private QRCodeZipService qrCodeZipService;
     @Resource
     private CookieSessionService cookieSessionService;
+    @Resource
+    private Sender sender;
     private final Logger logger = LoggerFactory.getLogger(TestControl.class);
 
+    @GetMapping("send")
+    public void send(@RequestParam String msg) {
+        sender.sendHttpRequest(msg);
+    }
+
     @LogInfo("redis功能测试")
     @GetMapping("redis")
     public void redis() {

boot-web/src/main/resources/application.yml

@@ -48,12 +48,12 @@ spring:
       force: true
   #消息队列
   rabbitmq:
-     host:
-     port:
-     username:
-     password:
+     host: 127.0.0.1
+     port: 5672
+     username: guest
+     password: guest
      virtual-host: /
-     connection-timeout:
+     connection-timeout: 5
      publisher-confirms: true
      publisher-returns: true
 #log