diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConfig.java b/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConfig.java index 971d835..dbf8445 100644 --- a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConfig.java +++ b/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); } } diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java b/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java index 720d6cd..645c10c 100644 --- a/boot-service/src/main/java/com/xwbing/Rabbit/RabbitConstant.java +++ b/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"; } diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/Receiver.java b/boot-service/src/main/java/com/xwbing/Rabbit/Receiver.java index f691565..64296d0 100644 --- a/boot-service/src/main/java/com/xwbing/Rabbit/Receiver.java +++ b/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); } } diff --git a/boot-service/src/main/java/com/xwbing/Rabbit/Sender.java b/boot-service/src/main/java/com/xwbing/Rabbit/Sender.java index a91b383..7ff1f39 100644 --- a/boot-service/src/main/java/com/xwbing/Rabbit/Sender.java +++ b/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); }