diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConfig.java b/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConfig.java new file mode 100644 index 0000000..971d835 --- /dev/null +++ b/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); + } +} diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java b/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java new file mode 100644 index 0000000..720d6cd --- /dev/null +++ b/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"; +} diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/Receiver.java b/boot-service/src/main/java/com/xwbing/Rabbit/Receiver.java new file mode 100644 index 0000000..f691565 --- /dev/null +++ b/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); + } +} diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/Sender.java b/boot-service/src/main/java/com/xwbing/Rabbit/Sender.java new file mode 100644 index 0000000..a91b383 --- /dev/null +++ b/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); + } +}