merge

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

@@ -0,0 +1,45 @@
+package com.xwbing.Rabbit;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * 项目名称: boot-module-demo
+ * 创建时间: 2018/4/25 14:28
+ * 作者: xiangwb
+ * 说明:
+ */
+@Configuration
+public class RabbitConfig {
+    //声明队列
+    @Bean
+    public Queue serverInvokeQueue() {
+        return new Queue(RabbitConstant.SERVER_INVOKE_QUEUE, true);//true表示持久化该队列
+    }
+
+    @Bean
+    public Queue httpRequestQueue() {
+        return new Queue(RabbitConstant.HTTP_REQUEST_QUEUE, true);
+    }
+
+    //声明交互器
+    @Bean
+    public TopicExchange topicExchange() {
+        return new TopicExchange(RabbitConstant.CONTROL_EXCHANGE);
+    }
+
+    //绑定
+    @Bean
+    public Binding bindingExchangeMessage() {
+        return BindingBuilder.bind(serverInvokeQueue()).to(topicExchange()).with(RabbitConstant.SERVER_INVOKE_ROUTINGKEY);
+    }
+
+    @Bean
+    public Binding bindingExchangeMessages() {
+        return BindingBuilder.bind(httpRequestQueue()).to(topicExchange()).with(RabbitConstant.HTTP_REQUEST_ROUTINGKEY);
+    }
+}

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

@@ -0,0 +1,15 @@
+package com.xwbing.Rabbit;
+
+/**
+ * 项目名称: boot-module-demo
+ * 创建时间: 2018/4/25 14:34
+ * 作者: xiangwb
+ * 说明:
+ */
+public class RabbitConstant {
+    public static final String SERVER_INVOKE_QUEUE = "server_invoke";
+    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 CONTROL_EXCHANGE = "control_exchange";
+}

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

@@ -0,0 +1,31 @@
+package com.xwbing.Rabbit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import static com.xwbing.Rabbit.RabbitConstant.HTTP_REQUEST_QUEUE;
+import static com.xwbing.Rabbit.RabbitConstant.SERVER_INVOKE_QUEUE;
+
+/**
+ * 项目名称: boot-module-demo
+ * 创建时间: 2018/4/25 15:12
+ * 作者: xiangwb
+ * 说明:
+ */
+@Component
+public class Receiver {
+    private final Logger logger = LoggerFactory.getLogger(Receiver.class);
+
+    @RabbitListener(queues = SERVER_INVOKE_QUEUE)
+    public String processServer(String msg) {
+        logger.info(Thread.currentThread().getName() + " 接收到来自{}队列的消息:{}", 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);
+    }
+}

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

@@ -0,0 +1,54 @@
+package com.xwbing.Rabbit;
+
+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.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.util.UUID;
+
+/**
+ * 项目名称: boot-module-demo
+ * 创建时间: 2018/4/25 14:48
+ * 作者: xiangwb
+ * 说明:
+ */
+@Component
+public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
+    @Resource
+    private RabbitTemplate rabbitTemplate;
+    private final Logger logger = LoggerFactory.getLogger(Sender.class);
+
+    @PostConstruct
+    public void init() {
+        rabbitTemplate.setConfirmCallback(this);
+        rabbitTemplate.setReturnCallback(this);
+    }
+
+    @Override
+    public void confirm(CorrelationData correlationData, boolean b, String cause) {
+        if (b) {
+            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) {
+        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();
+        logger.info("结束发送消息:{}", msg.toLowerCase());
+        logger.info("消费者响应:{}消息处理完成", response);
+    }
+}