merge

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

@@ -35,11 +35,11 @@ public class RabbitConfig {
     //绑定
     @Bean
     public Binding bindingExchangeMessage() {
-        return BindingBuilder.bind(serverInvokeQueue()).to(topicExchange()).with(RabbitConstant.SERVER_INVOKE_ROUTINGKEY);
+        return BindingBuilder.bind(serverInvokeQueue()).to(topicExchange()).with(RabbitConstant.SERVER_INVOKE_ROUTING_KEY);
     }
 
     @Bean
     public Binding bindingExchangeMessages() {
-        return BindingBuilder.bind(httpRequestQueue()).to(topicExchange()).with(RabbitConstant.HTTP_REQUEST_ROUTINGKEY);
+        return BindingBuilder.bind(httpRequestQueue()).to(topicExchange()).with(RabbitConstant.HTTP_REQUEST_ROUTING_KEY);
     }
 }

boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java

@@ -7,9 +7,24 @@ package com.xwbing.Rabbit;
  * 说明:
  */
 public class RabbitConstant {
+    /**
+     * 服务器调用
+     */
     public static final String SERVER_INVOKE_QUEUE = "server_invoke";
+    /**
+     * http请求
+     */
     public static final String HTTP_REQUEST_QUEUE = "http_request";
-    public static final String SERVER_INVOKE_ROUTINGKEY = "server_invoke.key";
-    public static final String HTTP_REQUEST_ROUTINGKEY = "http_request.key";
+    /**
+     * 服务器调用路由键
+     */
+    public static final String SERVER_INVOKE_ROUTING_KEY = "server_invoke.key";
+    /**
+     * http请求路由键
+     */
+    public static final String HTTP_REQUEST_ROUTING_KEY = "http_request.key";
+    /**
+     * 交换器
+     */
     public static final String CONTROL_EXCHANGE = "control_exchange";
 }

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

@@ -20,12 +20,12 @@ public class Receiver {
 
     @RabbitListener(queues = SERVER_INVOKE_QUEUE)
     public String processServer(String msg) {
-        logger.info(Thread.currentThread().getName() + " 接收到来自{}队列的消息:{}", SERVER_INVOKE_QUEUE, msg);
+        logger.info("接收到来自{}队列的消息:{}", SERVER_INVOKE_QUEUE, msg);
         return msg.toUpperCase();
     }
 
     @RabbitListener(queues = HTTP_REQUEST_QUEUE)
     public void processHttp(String msg) {
-        logger.info(Thread.currentThread().getName() + " 接收到来自{}队列的消息:{}", HTTP_REQUEST_QUEUE, msg);
+        logger.info("接收到来自{}队列的消息:{}", HTTP_REQUEST_QUEUE, msg);
     }
 }

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

@@ -25,29 +25,53 @@ public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.Re
 
     @PostConstruct
     public void init() {
+        //用于实现消息发送到RabbitMQ交换器后接收ack回调。
         rabbitTemplate.setConfirmCallback(this);
+        //用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
         rabbitTemplate.setReturnCallback(this);
     }
 
+    /**
+     * 相应交换器接收后回调
+     */
     @Override
-    public void confirm(CorrelationData correlationData, boolean b, String cause) {
-        if (b) {
+    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+        if (ack) {
             logger.info("消息发送成功:{}", correlationData);
         } else {
             logger.info("消息发送失败:{}", cause);
         }
     }
 
+    /**
+     * 无交换器绑定回调
+     */
     @Override
     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
         logger.info(message.getMessageProperties().getCorrelationIdString() + "发送失败");
     }
 
-    //发送消息，不需要实现任何接口，供外部调用。
-    public void send(String msg) {
+    public void sendServerInvoke(String msg) {
+        send(msg, RabbitConstant.CONTROL_EXCHANGE, RabbitConstant.SERVER_INVOKE_ROUTING_KEY);
+    }
+
+    public void sendHttpRequest(String msg) {
+        send(msg, RabbitConstant.CONTROL_EXCHANGE, RabbitConstant.HTTP_REQUEST_ROUTING_KEY);
+    }
+
+
+    /**
+     * 发送消息
+     *
+     * @param msg        消息
+     * @param exchange   交换器
+     * @param routingKey 路由键
+     */
+    private void send(String msg, String exchange, String routingKey) {
         CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
         logger.info("开始发送消息:{}", msg.toLowerCase());
-        String response = rabbitTemplate.convertSendAndReceive(RabbitConstant.CONTROL_EXCHANGE, RabbitConstant.SERVER_INVOKE_ROUTINGKEY, msg, correlationId).toString();
+        //转换并发送消息,且等待消息者返回响应消息。
+        String response = rabbitTemplate.convertSendAndReceive(exchange, routingKey, msg, correlationId).toString();
         logger.info("结束发送消息:{}", msg.toLowerCase());
         logger.info("消费者响应:{}消息处理完成", response);
     }