RabbitMQ --- 直连交换机 【 同步操作，等到消费者处理完后返回处理结果 】

1.前言

　　RabbleMQ这种消息中间件，主要的功能是使用异步操作，来达到解耦的目的，比如，有系统a和系统b,

系统a通过消息中间件通知系统b来做业务，那么系统a只需要把要做的事情【也就是消息】发给消息中间件后，

消息中间件就会把消息转发给系统b，系统a不需要关心系统b是怎么完成业务的，也不需要关心业务完成的结果，

这是就是异步操作。

　　如果系统a希望获得系统b的处理结果，那么系统a使用消息中间件发送消息后需要原地等待，做阻塞操作，但是

等待时长不能超过最大超时时间，可设置RabbleMQ自定义超时时间，这样还不如直接调用该业务呢，何必再加个消息消息中间件通知他来做？

　　所以，一般不会做这样的同步阻塞操作，违背了消息中间件的开发初衷【提高吞吐量和系统业务的响应速度】，虽然

不影响解耦度，但是这样的操作使得消息中间件变得不伦不类了。

　　那么要问了，使用RabbleMQ消息中间件的具体好处是什么？

　　第一，解耦，是非常明显的好处，通过消息中间件，系统a通知系统b做什么业务，不需要关心系统b是怎么实现业务的。
　　第二，异步，非必要的业务，不需要特别关心结果的业务，可以写入消息中间件以异步方式操作，横向编程，加快响应速度。
　　第三，削峰，并发量大的时候直接将所有数据怼到数据库，数据库会异常的，使用消息队列的特性，所有的操作会进行排队，
　　　　消息也不会丢失，当消息被消费后才会销毁，比如秒杀系统就是这样实现的。

　　那么，虽然同步阻塞操作偶很大的闭端，那么到底该怎么操作？

这篇随笔 以 随笔 https://www.cnblogs.com/c2g5201314/p/13156932.html 为基础 ，修改部分代码实现演示

2.操作

（1）修改消息生产者的消息生产类的 rabbitTemplate 模板发送消息方法 ，同步则使用 convertSendAndReceive（），参数与异步非阻塞操作的一样。

 

 源码

package com.example.rabbitmqproducer1004.rabbitmqFactory;


import com.example.rabbitmqproducer1004.config.RabbitmqConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * 消息生产类
 */
@Component
//实现接口
public class SendMessage  {

//需要设置回调方法，获取消费结果才需要实现 RabbitTemplate.ConfirmCallback 接口，
//public class SendMessage implements RabbitTemplate.ConfirmCallback {
    Logger logger = LoggerFactory.getLogger(this.getClass());

//======================================================================
//    /**
//     * 方法一：设置回调方法，获取消费结果，
//     * <p>
//     * 缺点是：必须手动配置RabbitTemplate模板 ，代码量大
//     */
//
//    //存储 rabbitmq模板的临时变量
//    private final RabbitTemplate rabbitTemplate;
//
//    /**
//     * 构造注入rabbitmq模板,这样可以设置回调方法，获取消费结果，但是必须手动配置RabbitTemplate模板
//     */
//    @Autowired
//    public SendMessage(RabbitTemplate rabbitTemplate) {
//        this.rabbitTemplate = rabbitTemplate;
//        //设置确认回调的方法，参数类型为ConfirmCallback
//        this.rabbitTemplate.setConfirmCallback(this);
//    }
//
//    /**
//     * 回调方法，获取消费结果
//     *
//     * @param correlationData 关联数据
//     * @param b               消息是否被消费成功，成功为true ,失败为false
//     * @param s               原因 ，消费成功则返回null,否则返回失败原因
//     */
//    @Override
//    public void confirm(CorrelationData correlationData, boolean b, String s) {
//        logger.warn("回调的连接数据：" + correlationData);
//        if (correlationData != null) {
//            //CorrelationData [id=1bcab025-2b4c-4f74-a22d-41007e30f551]
//            logger.warn("获取correlationData的id值:" + correlationData.getId());
//        }
//        //1bcab025-2b4c-4f74-a22d-41007e30f551
//        if (b) {
//            logger.warn("回调结果：消息消费成功");
//        } else {
//            logger.warn("回调结果：失败。原因：" + s);
//        }
//    }

//========================================================================
//    /**
//     * 方法二 :不需要获取获取消费结果，只需要发送即可
//     *
//     * 优点：自动装配，代码量少
//     */
    @Autowired
    private RabbitTemplate rabbitTemplate;

//========================================================================

    /**
     * 发送消息
     * <p>
     * 参数是消息内容
     */
    public void send(String message) {
        logger.warn("发送消息，内容：" + message);
        /**
         * 方法一：异步操作，不等待消费者端返回处理结果，设置在回调操作的关联数据，用于识别是哪一条消息和确认是否执行成功
         */
////        实例关联数据对象，使用UUID随机数 作为 回调id
//        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
////        发送消息 ，参数分别是 : 指定的交换机名字 、指定的路由关键字、消息字符串、关联数据
//        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANG_1,RabbitmqConfig.ROUTINGKEY_1,message,correlationData);
        /**
         * 方法二：异步操作，不等待消费者端返回处理结果，且在消息回调操作的关联数据为null，如果不做回调操作，则建议这样使用
         */
//        //发送消息 ，参数分别是 : 指定的交换机名字 、指定的路由关键字、消息字符串
//        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANG_1, RabbitmqConfig.ROUTINGKEY_1, message);
        /**
         * 方法三：同步操作，等待消费者端返回处理结果
         */
        Object dd = rabbitTemplate.convertSendAndReceive(RabbitmqConfig.EXCHANG_1, RabbitmqConfig.ROUTINGKEY_1, message);
        logger.warn("你大爷的，有么有延迟？是不是同步阻塞？");
        logger.warn("结果是什么？？？==" + dd);


    }
}

 （2）修改消息消费者的监听方法 ，可设置任意类型返回值，但是在生产者端需要解析，我这里使用 字符串，休眠3秒

 源码

package com.example.rabbitmqconsumer1002.rabbitmqListener;

import com.example.rabbitmqconsumer1002.config.RabbitConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消息监听类--发短信
 */
//注册bean
@Component
//设置需要监听的消息队列
@RabbitListener(queues = RabbitConfig.QUEUE_1)
public class SendMessageListener {
    Logger logger = LoggerFactory.getLogger(getClass());



    //消息事件处理--有返回结果
    @RabbitHandler
    public String sendMessage(String msg) throws InterruptedException {
        logger.warn("我是端口1002的消费者,收到信息：" + msg);
        logger.warn("休眠3秒");
        Thread.sleep(3000);
        logger.warn("休眠结束");
        return "发送成功，是同步阻塞的么？";
    }
}

（3）启动工程

访问网址     http://localhost:1004/mq?msg=你大爷，帮我发短信3999

查看消费者端控制台打印

 查看生产者端控制台打印

 

 生产者端等待了3秒后才收到结果

3.如果换成10秒会怎么样？

 修改消息消费者的监听方法

 启动工程后 访问 网址  http://localhost:1004/mq?msg=你大爷，帮我发短信3999

查看消费者端控制台打印

 

  查看生产者端控制台打印

 打印控制台源码

2020-06-19 02:40:17.100 WARN  http-nio-1004-exec-4 | com.example.rabbitmqproducer1004.rabbitmqFactory.SendMessage | 发送消息，内容：你大爷，帮我发短信3999
2020-06-19 02:40:17.111 INFO  http-nio-1004-exec-4 | org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler | Initializing ExecutorService
2020-06-19 02:40:17.117 INFO  http-nio-1004-exec-4 | org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer | Container initialized for queues: [amq.rabbitmq.reply-to]
2020-06-19 02:40:17.128 INFO  http-nio-1004-exec-4 | org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer | SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-pPnAaXXfnkytocGRd7ES5Q identity=7ba94c9e] started
2020-06-19 02:40:22.137 WARN  http-nio-1004-exec-4 | com.example.rabbitmqproducer1004.rabbitmqFactory.SendMessage | 你大爷的，有么有延迟？是不是同步阻塞？
2020-06-19 02:40:22.137 WARN  http-nio-1004-exec-4 | com.example.rabbitmqproducer1004.rabbitmqFactory.SendMessage | 结果是什么？？？==null
2020-06-19 02:40:27.158 WARN  pool-1-thread-4 | org.springframework.amqp.rabbit.core.RabbitTemplate | Reply received after timeout for 1
2020-06-19 02:40:27.163 WARN  pool-1-thread-4 | org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler | Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1651) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1555) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1478) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1466) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1461) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1410) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.callExecuteListener(DirectMessageListenerContainer.java:996) [spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.handleDelivery(DirectMessageListenerContainer.java:956) [spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149) [amqp-client-5.4.3.jar:5.4.3]
    at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104) [amqp-client-5.4.3.jar:5.4.3]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_221]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_221]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_221]
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: Reply received after timeout
    at org.springframework.amqp.rabbit.core.RabbitTemplate.onMessage(RabbitTemplate.java:2535) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.lambda$setMessageListener$1(DirectReplyToMessageListenerContainer.java:115) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1552) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    ... 11 common frames omitted
2020-06-19 02:40:27.164 ERROR pool-1-thread-4 | org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer | Failed to invoke listener
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1651) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1555) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1478) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1466) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1461) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1410) ~[spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.callExecuteListener(DirectMessageListenerContainer.java:996) [spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.handleDelivery(DirectMessageListenerContainer.java:956) [spring-rabbit-2.1.7.RELEASE.jar:2.1.7.RELEASE]
    at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149) [amqp-client-5.4.3.jar:5.4.3]
    at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104) [amqp-client-5.4.3.jar:5.4.3]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_221]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_221]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_221]
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: Reply received after timeout

 抛出了监听异常，显然是等待超时了，然后继续执行执行下面的操作，返回结果以null处理

4.分析与思考

如果等待10分钟才能运行完这个业务，使用同步阻塞操作，岂不是需要等10分钟？那还要消息中间件有什么用？

因此，一般是使用异步非阻塞操作消息中间件的，因此，会尽可能的把一些不需要返回结果的操作用消息中间件转发完成。

对于失败和抛出异常的操作，可以制作一个死信队列，将这些无法完成的消息重写放入死信队列，让专门的系统监听进行处理，一般这样的业务很少出现，基本上是可以成功完成的，

对那些重要的，必须获取处理结果的业务，直接调用实现就好了，因为这样是默认同步阻塞操作的，不需要再转一趟消息中间件啦。