[大数据]（RabbitMQ快速上手）一文搞懂RabbitMQ全部工作模式！！普通模式，工作模式， 发布订阅模式， Direct路由模式，Topic主题模式，定时队列与死信队列，可靠生成和消费

前言

注：
大家好我是妈妈的好大儿，
笔者联系方式
QQ：3302254385
微信：yxc3302254385
交个朋友！
创作不易，三连十分感谢！！！

简介

本篇博文将实际代码的方式结合图片的方式演示常用的，rabbitMQ的模式！！按下面列表逐一演示，按需自取，总的配置文件讲放在文章最后！！！

1. 普通模式
2. 工作模式
3. 发布订阅模式
4. Direct路由模式
5. Topic主题模式
6. 发送定时消息
7. 定时队列与死信队列
8. 可靠生成和消费

Code

1.普通模式

简单模式是最简单的消息模式，它包含一个生产者、一个消费者和一个队列。生产者向队列里发送消息，消费者从队列中获取消息并消费

定义：

  	/**
     * 普通队列1
     */
    public static final String DEFAULT_QUEUE1_TO_TEST ="defaultQueue1ToTest";

创建队列：

    //1.普通队列1
    @Bean(name = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
    public Queue createQueue1Test(){
        /*
        参数：
        name--->队列名称-不能为null,设置为""可以使代理生成名称。
        durable--->如果我们声明一个持久队列，则为true（该队列将在服务器重启后继续存在）
        Exclusive---> 如果我们声明一个排他队列，则为true（该队列仅由声明者的连接使用）
        autoDelete--->如果服务器在不再使用队列时应该删除队列，则为true
        arguments---> 用于声明队列的参数
         */
        return new Queue(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,true,false,false,null);
    }

发送默认消息：

  	/**
     * 1.发送Defalut消息到队列
     * @return
     */
    @GetMapping("/sendDefalutMsgToQueue")
    public String sendDefalutMsgToQueue(){
        rabbitTemplate.convertAndSend(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,"普通消息");
        return "true";
    }

消费者：

@Component
//该方法将方法标记为指定queues() （或bindings() ）上的Rabbit消息侦听器的目标
@RabbitListener(queues = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
public class MsgDefaultConsumer1 {

    //该方法将方法标记为用RabbitListener注释的类中的Rabbit消息侦听器的目标
    @RabbitHandler
    public void getMsg(String msg){
        System.out.println("收到了defaultQueue1ToTest的消息--->"+msg);
    }

}

结果：

发送几次收到几条

2.工作模式

工作模式是指向多个互相竞争的消费者发送消息的模式，它包含一个生产者、两个消费者和一个队列。两个消费者同时绑定到一个队列上去，当消费者获取消息处理耗时任务时，空闲的消费者从队列中获取并消费消息。

就是一个队列多个消费，一条消息只能被一个消费者消费

定义：

  	/**
     * 普通队列1
     */
    public static final String DEFAULT_QUEUE1_TO_TEST ="defaultQueue1ToTest";

创建队列：

    //1.普通队列1
    @Bean(name = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
    public Queue createQueue1Test(){
        /*
        参数：
        name--->队列名称-不能为null,设置为""可以使代理生成名称。
        durable--->如果我们声明一个持久队列，则为true（该队列将在服务器重启后继续存在）
        Exclusive---> 如果我们声明一个排他队列，则为true（该队列仅由声明者的连接使用）
        autoDelete--->如果服务器在不再使用队列时应该删除队列，则为true
        arguments---> 用于声明队列的参数
         */
        return new Queue(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,true,false,false,null);
    }

发送工作消息：

   /**
     * 2.发送work消息到队列
     * @return
     */
    @GetMapping("/sendWorkMsgToQueue")
    public String sendWorkMsgToQueue(){
        for (int i = 0; i <20 ; i++) {
            //正常发送 
            rabbitTemplate.convertAndSend(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,"工作模式消息");
            //公平分发
            //rabbitTemplate.convertSendAndReceive(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,"工作模式消息");
            
        }

        return "true";
    }

消费者：

@Component
//该方法将方法标记为指定queues() （或bindings() ）上的Rabbit消息侦听器的目标
@RabbitListener(queues = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
public class MsgDefaultConsumer1 {

    //该方法将方法标记为用RabbitListener注释的类中的Rabbit消息侦听器的目标
    @RabbitHandler
    public void getMsg(String msg){
        System.out.println("收到了defaultQueue1ToTest的消息--->"+msg);
    }

}

-------------------------------------------------------------------------------------------
@Component
public class MsgWorkConsumer2 {

    @RabbitListener(queues = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
    public void getMsg(String msg){
        System.out.println("Consumer2--->收到了defaultQueue1ToTest的消息--->"+msg);
    }

}

一.正常发送，谁的能力更强谁的消费条数就越多

二.公平分发 按照消费者的顺序来，性能特别慢

3.发布订阅模式

一条消息通过交换机直接发送到2个订阅的队列

一条消息发送到交换机，只要与交换机绑定关系的（也就相当于队列订阅了交换机）每个队列都将收到消息**

定义：

    /**
     * 发布订阅模式队列1
     */
    public static final String PUBLISH_QUEUE1_TO_TEST="publish_Queue1ToTest";

    /**
     * 发布订阅模式队列1
     */
    public static final String PUBLISH_QUEUE2_TO_TEST="publish_Queue2ToTest";

    /**
     * 发布订阅fanout模式交换机
     */
    public static final String FANOUT_EXCHANGE= "fanoutExchange";

创建队列交换机：

   //2.发布订阅模式队列1
    @Bean(name = RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST)
    public Queue createPublishQueue1Test(){
        return new Queue(RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST,true,false,false,null);
    }

    //2.发布订阅模式队列2
    @Bean(name = RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST)
    public Queue createPublishQueue2Test(){
        return new Queue(RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST,true,false,false,null);
    }

    //2.发布订阅模式交换机
    @Bean(name = RabbitMQBeanName.FANOUT_EXCHANGE)
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("exchange.fanout",true,false,null);
    }

    //2.发布订阅模式Binding
    @Bean
    public Binding fanoutBinding1(@Qualifier(RabbitMQBeanName.FANOUT_EXCHANGE) FanoutExchange fanout, @Qualifier(RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST)Queue fanoutQueue1) {
        return BindingBuilder.bind(fanoutQueue1).to(fanout);
    }

    //将队列二绑定到交换机
    @Bean
    public Binding fanoutBinding2(@Qualifier(RabbitMQBeanName.FANOUT_EXCHANGE) FanoutExchange fanout, @Qualifier(RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST)Queue fanoutQueue2) {
        return BindingBuilder.bind(fanoutQueue2).to(fanout);
    }

发送消息：

  /**
     * 3.fanout模式
     */
    @GetMapping("/sendFanoutMsgToQueue")
    public String sendFanoutMsgToQueue(){
        rabbitTemplate.convertAndSend(RabbitMQBeanName.FANOUT_EXCHANGE,"","发布订阅模式消息");
        return "true";
    }

消费者：

/**
 * @Author: Joker-CC
 * @Path:
 * @Date 2021/06/04 14:46
 * @Description: 发布订阅消费者
 * @Version: 1.0
 */
@Component
public class MsgPublishConsumer3 {

    @RabbitListener(queues = RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST)
    public void consumer1(String msg){
      System.out.println("Consumer1--->收到了PUBLISH_QUEUE1_TO_TEST的消息--->"+msg);
    }

    @RabbitListener(queues = RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST)
    public void consumer2(String msg){
        System.out.println("Consumer2--->收到了PUBLISH_QUEUE2_TO_TEST的消息--->"+msg);
    }


}

4.Direct路由模式

发送消息到交换机，交换机根据路由key，路由到指定的队列，消费者监听队列进行消费

定义：

 /**
     * 发布订阅模式队列1
     */
    public static final String PUBLISH_QUEUE1_TO_TEST="publish_Queue1ToTest";

    /**
     * 发布订阅模式队列2
     */
    public static final String PUBLISH_QUEUE2_TO_TEST="publish_Queue2ToTest";

    /**
     * 发布订阅fanout模式交换机
     */
    public static final String FANOUT_EXCHANGE= "fanoutExchange";

创建队列交换机并绑定关系：

 /**
     * 3.创建Direct交换机
     * @return
     */
    @Bean(name = RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST)
    public DirectExchange createDirectExchangeTest(){
       /*
        参数说明：
        name--->交换机的名称。
        durable--->是否持久化-如果我们声明一个持久交换机（为true该持久交换机将在服务器重新启动后继续存在）
        autoDelete--->如果服务器在不再使用该交换时应删除该交换机
        arguments--->用于声明交换的参数
         */
        return new DirectExchange(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST,true,false,null);
    }

    //3.Direct队列1
    @Bean(name = RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST)
    public Queue createDirectQueue1Test(){
        return new Queue(RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST,true,false,false,null);
    }

    //3.Direct队列1
    @Bean(name = RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST)
    public Queue createDirectQueue2Test(){
        return new Queue(RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST,true,false,false,null);
    }

    //3.路由模式Binding1
    @Bean
    public Binding directBinding1(@Qualifier(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST) DirectExchange direct, @Qualifier(RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST) Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(direct).with("sb");
    }

    //3.路由模式Binding2
    @Bean
    public Binding directBinding2(@Qualifier(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST) DirectExchange direct, @Qualifier(RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST) Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(direct).with("wc");
    }

消息发送：

 /**
     * 4.Direct模式
     */
    @GetMapping("/sendDirectMsgToQueue")
    public String sendDirectMsgToQueue(){
        rabbitTemplate.convertAndSend(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST,"sb","发布路由模式消息");
        return "true";
    }

消息消费：

/**
 * @Author: Joker-CC
 * @Path:
 * @Date 2021/06/04 18:02
 * @Description: 路由模式消费者
 * @Version: 1.0
 */
@Component
public class MsgDirectConsumer4 {

    @RabbitListener(queues = RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST)
    public void consumer1(String msg){
        System.out.println("Consumer1--->收到了DIRECT_QUEUE1_TO_TEST的消息--->"+msg);
    }

    @RabbitListener(queues = RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST)
    public void consumer2(String msg){
        System.out.println("Consumer2--->收到了DIRECT_QUEUE2_TO_TEST的消息--->"+msg);
    }



}

结果：

发送2次只有队列1的路由key匹配上，就会收到2条消息

5.Topic主题模式

路由模式的升级版加了通配符 ，消息发送时指定routingKey和队列的routingKey进行匹配，匹配成功则向队列传递消息

routingKey说明

• #号代表，匹配0个或多个
• *号代表，匹配一个

我们在创建队列时，会有一个routingKey—>路由key

定义：

  /**
     * topic模式队列1
     */
    public static final String TOPIC_QUEUE1_TO_TEST ="topicQueue1ToTest";

    /**
     * topic模式队列2
     */
    public static final String TOPIC_QUEUE2_TO_TEST ="topicQueue2ToTest";

    /**
     * topic模式交换机
     */
    public static final String TOPIC_EXCHANGE_TO_TEST="topicExchangeToTest";

创建交换机和队列并绑定关系：

/**
     * 4.创建topic交换机
     * @return
     */
    @Bean(name = RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST)
    public TopicExchange createTopicExchangeTest(){
       /*
        参数说明：
        name--->交换机的名称。
        durable--->是否持久化-如果我们声明一个持久交换机（为true该持久交换机将在服务器重新启动后继续存在）
        autoDelete--->如果服务器在不再使用该交换时应删除该交换机
        arguments--->用于声明交换的参数
         */
        return new TopicExchange(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST,true,false,null);
    }

    //4.topic队列1
    @Bean(name = RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST)
    public Queue createTopicQueue1Test(){
        return new Queue(RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST,true,false,false,null);
    }

    //4.topic队列2
    @Bean(name = RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST)
    public Queue createTopicQueue2Test(){
        return new Queue(RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST,true,false,false,null);
    }

    //4.主题模式Binding1
    @Bean
    public Binding topicBinding1(@Qualifier(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST) TopicExchange topic, @Qualifier(RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST) Queue topicQueue1) {
        //通配符路由key
        return BindingBuilder.bind(topicQueue1).to(topic).with("#.wo.shi.sb.#");
    }

    //4.主题模式Binding2
    @Bean
    public Binding topicBinding2(@Qualifier(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST) TopicExchange topic, @Qualifier(RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST) Queue topicQueue2) {
        //通配符路由key
        return BindingBuilder.bind(topicQueue2).to(topic).with("*.wo.shi.sb.*");
    }  

发送消息：

        /**
     * 5.topic模式
     */
    @GetMapping("/sendTopicMsgToQueue")
    public String sendTopicMsgToQueue(){
        //此消息2个队列都将收到
        rabbitTemplate.convertAndSend(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST,"aa.wo.shi.sb.bb","发布主题模式消息1");
        //此消息只有队列1能收到
        rabbitTemplate.convertAndSend(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST,"aa.wo.shi.sb.bb.cc","发布主题模式消息1");
        return "true";
    }

消费者消费消息：

/**
 * @Author: Joker-CC
 * @Path:
 * @Date 2021/06/09 16:16
 * @Description: 主题模式消费者
 * @Version: 1.0
 */
@Component
public class MsgTopicConsumer5 {

    @RabbitListener(queues = RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST)
    public void consumer1(String msg){
        System.out.println("Consumer1--->收到了TOPIC_QUEUE1_TO_TEST的消息--->"+msg);
    }

    @RabbitListener(queues = RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST)
    public void consumer2(String msg){
        System.out.println("Consumer2--->收到了TOPIC_QUEUE2_TO_TEST的消息--->"+msg);
    }


}

结果：

发送2次

队列1收到4条

队列2收到2条

6.发送定时消息

定时消息如果没有被消费，时间到期了就删除了，不会进入死信队列

定义：

创建交换机和队列并绑定关系：

发送消息：

 /**
     * 6.发送定时消息到队列，消息过期后不会放入死信队列
     * @return
     */
    @GetMapping("/sendTTLMsgToQueue")
    public String sendTTLMsgToQueue(){
        String msg ="定时消息";

        MessagePostProcessor messagePostProcessor= new MessagePostProcessor(){
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //配置消息的过期时间 为10000毫秒  10秒
                message.getMessageProperties().setExpiration("10000");
                //配置消息编码格式
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };

        rabbitTemplate.convertAndSend(RabbitMQBeanName.FANOUT_EXCHANGE,"",msg,messagePostProcessor);
        return "true";
    }    

消费者消费消息：

结果：

可以看到消息保存了10秒没有被消费者消费就直接丢弃了

7.定时队列与死信队列

消息放入定时队列中，每条消息都有对应的过期时间，如果绑定了死信交换机，当消息过期时就会根据
路由key，发送到死信队列！！

定义：

    /**
     * 定时队列
     */
    public static final String TTL_QUEUE_TO_TEST ="ttlQueueToTest";


    /**
     * 死信交换机
     */
    public static final String DEAD_LETTER_EXCHANGE="deadLetterExchange";

    /**
     * 死信队列
     */
    public static final String DEAD_LETTER_QUEUE_TO_TEST="deadLetterQueueToTest";

创建交换机和队列并绑定关系：

     //5.创建定时队列
    @Bean(name = RabbitMQBeanName.TTL_QUEUE_TO_TEST)
    public Queue createTTLQueueTest(){
        /*
        参数：
        name--->队列名称-不能为null,设置为""可以使代理生成名称。
        durable--->如果我们声明一个持久队列，则为true（该队列将在服务器重启后继续存在）
        Exclusive---> 如果我们声明一个排他队列，则为true（该队列仅由声明者的连接使用）
        autoDelete--->如果服务器在不再使用队列时应该删除队列，则为true
        arguments---> 用于声明队列的参数
         */
        HashMap arguments = new HashMap<>(4);
        //配置延迟队列参数 标注这是一个延迟队列且消息5秒后删除，放入到死信队列
        arguments.put("x-message-ttl",5000);
        //绑定死信交换机
        arguments.put("x-dead-letter-exchange", RabbitMQBeanName.DEAD_LETTER_EXCHANGE);
        //绑定死信交换机路由key
        arguments.put("x-dead-letter-routing-key", "dead");
        return new Queue(RabbitMQBeanName.TTL_QUEUE_TO_TEST,true,false,false,arguments);
    }


    //5.创建死信交换机
    @Bean(RabbitMQBeanName.DEAD_LETTER_EXCHANGE)
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(RabbitMQBeanName.DEAD_LETTER_EXCHANGE,true,false,null);
    }

    //5.创建死信队列
    @Bean(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST)
    public Queue createDeadLetterQueue1Test(){
        return new Queue(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST,true,false,false,null);
    }

    //5.死信队列和死信交换机绑定
    @Bean
    public Binding deadLetterBindingB(@Qualifier(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST) Queue queue,
                                      @Qualifier(RabbitMQBeanName.DEAD_LETTER_EXCHANGE) DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("dead");
    }

发送消息：

       /**
     * 发送消息到定时队列
     * @return
     */
    @GetMapping("/sendMsgToTTLQueue")
    public String sendMsgToTTLQueue(){
        rabbitTemplate.convertAndSend(RabbitMQBeanName.TTL_QUEUE_TO_TEST,"定时队列消息");
        return "true";
    }

消费者消费消息：

/**
 * @Author: Joker-CC
 * @Path:
 * @Date 2021/06/09 17:53
 * @Description: 死信队列消费者
 * @Version: 1.0
 */
@Component
public class MsgTTLQueueDeadQueueConsumer6 {

    @RabbitListener(queues = RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST)
    public void getMsg(String msg){
        System.out.println("Consumer1--->收到了DEAD_LETTER_QUEUE_TO_TEST的消息--->"+msg);
    }

}

结果：

8.可靠生产和可靠消费

所谓的可靠生成就是从2个层面，第一是消息是否投递到了交换机！！！第二是消息是否被成功消费！！！

配置rabbitmqTemplate

/**
     * 自定义rabbitTemplate
     * @param connectionFactory
     * 它会建立一个可由应用程序共享的连接代理。共享连接是可能的，因为使用 AMQP 进行消息传递的“工作单元”实际上是一个“通道”（在某些方面，这类似于 JMS 中连接和会话之间的关系）。
     * 连接实例提供了一种createChannel方法。该CachingConnectionFactory实现支持对这些通道进行缓存，并根据通道是否为事务性维护单独的缓存。创建 的实例时CachingConnectionFactory，您可以通过构造函数提供“主机名”。
     * 您还应该提供“用户名”和“密码”属性。要配置通道缓存的大小（默认为 25），可以调用该 setChannelCacheSize()方法
     * @return
     */
    @Bean("diyRabbitTemplate")
    public RabbitTemplate initAckRabbitTemplate(@Autowired CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
        rabbitTemplate.setMandatory(true);
        //为此模板设置消息转换器。 用于将 Object 参数解析为 convertAndSend 方法和来自 receiveAndConvert 方法的 Object 结果。默认转换器是 SimpleMessageConverter，它能够根据消息内容类型标头处理字节数组、字符串和可序列化对象。
        rabbitTemplate.setMessageConverter(converter());

        //----------------------------------------发布者重试机制-------------------------------------------------------------
        /*
        消息是否成功发送到Exchange
        参数：
            correlationData相关数据 – 回调的相关数据。
            ack – ack 为真，nack 为假
            cause 原因 – 可选原因，如果可用，则为 nack，否则为 null。
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                String msgId = correlationData.getId();
                log.info("消息成功发送到Exchange--->"+msgId);
                //msgLogService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_SUCCESS);
            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });


        /*
        消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
        参数：
            message – 返回的消息。
            回复代码 - 回复代码。
            回复文本 – 回复文本。
            交换——交换。
            routingKey – 路由密钥。
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });

        return rabbitTemplate;
    }

    public Jackson2JsonMessageConverter converter() {
        return new Jackson2JsonMessageConverter();
    }

配置消费者：

  #开启发布者回调
    publisher-returns: true
    # 发送确认  具体配置可以看此类 ConfirmType
    publisher-confirm-type: correlated
    #设置手动ack
    listener:
      simple:
        #手动确认 具体配置可以看此类 AcknowledgeMode
        acknowledge-mode: MANUAL
        #每个消费者可以未确认的最大未确认消息数  具体配置可以看此类AmqpContainer具体实现类是SimpleContainer
        prefetch: 100
        #Spring官网的列子使用 retryTemplate进行配置
        retry:
          #开启发布重试 (默认false)
          enabled: true
          #传递消息的最大尝试次数 (默认3)
          max-attempts: 2
          #第一次和第二次尝试传递消息之间的持续时间  重试间隔时间（单位毫秒 默认1000ms）
          initial-interval: 5000ms
          #尝试之间的最长持续时间 （默认10000ms）
          max-interval: 20000ms
          #应用于前一个重试间隔的乘数。 （默认1.0） 间隔时间*乘数=下一次的间隔时间，最大不能超过设置的最大间隔时间
          multiplier: 1

配置生产者：

  #开启发布者回调
    publisher-returns: true
    # 发送确认  具体配置可以看此类 ConfirmType
    publisher-confirm-type: correlated
    #设置手动ack
    listener:
      simple:
        #手动确认 具体配置可以看此类 AcknowledgeMode
        acknowledge-mode: MANUAL

生产者

    /**
     * 发送可靠消息
     * @return
     */
    @GetMapping("/sendAckMsg")
    public String sendAckMsg(){
        //消息id
        String msgId = UUID.randomUUID().toString();
        //用于将发布者确认与发送的消息相关联的基类。 使用包含其中之一作为参数的org.springframework.amqp.rabbit.core.RabbitTemplate方法； 当收到发布者确认时，CorrelationData 与 ack/nack 一起返回。
        //使用提供的 id 构造一个实例。群必须是唯一的returnedMessage 。
        CorrelationData correlationData = new CorrelationData(msgId);
        //参数：1.交换机  2.路由key  3.msg  4.correlationData
        diyRabbitTemplate.convertAndSend(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST,"sb","ack消息-->"+msgId,correlationData);
        return "true";
    }

消费者：

/**
     * 保证消息能够成功消费
     * 注：
     * 1.如果是抛出异常就按照yaml配置文件里的重试机制配置的进行重试
     * 2.如果是使用 channel.basicNack(tag, false, true); 会一直进行重试导致内存溢出
     * 3.如果是使用 channel.basicNack(tag, false, false);会进入死信队列 删除消息
     * @param msg
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST)
    public void consumer1(Message msg, Channel channel) throws IOException {

        MessageProperties properties = msg.getMessageProperties();
        long tag = properties.getDeliveryTag();
        System.out.println("Consumer1--->收到了DIRECT_QUEUE1_TO_TEST的消息--->"+msg.toString());

//        /**
//         * 确认收到的一条或多条消息
//         * @param  DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
//         * @param  多个 – true 确认所有消息，包括提供的交付标签； false 仅确认提供的交付标签
//         */
//        channel.basicAck(tag, true);

        throw new RuntimeException();



        //消息消费失败
        /**
         * 拒绝收到的一个或多个消息
         * 参数：
         * DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
         * 多个 – true 拒绝所有消息，包括提供的交付标签； false 仅拒绝提供的交付标签。
         * requeue – 如果被拒绝的消息应该重新排队而不是丢弃则为true 否则反之进入死信，
         */
//        channel.basicNack(tag, false, false);
        
        
        
    }

注：

• 发送者发送可靠消息
  • 第一配置可靠生产
  • 第二如果需要发送到死信队列配置，死信交换机和队列
  • 第三发送消息时添加一个CorrelationData correlationData = new CorrelationData(msgId); 生成一个唯一的msgid用于确认消息或拒绝消息！！！
• 生产者确认消息
  • 第一配置可靠消费，重试次数，重试时间间隔
  • 第二确认消息
    • 消息确认接收 channel.basicAck(tag, true);
    • 消息拒绝 拒绝交付重新进入队列会无限发送（发送次数和配置无关） 最后进入死信队列 channel.basicNack(tag, false, true);
    • 消息拒绝 拒绝交付重新进入死信队列 channel.basicNack(tag, false, flase);
    • 抛出异常会重试 按照配置文件的里面的

完整配置文件

配置类

package com.cc.config;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.impl.AMQImpl;
import com.sun.org.apache.bcel.internal.generic.NEW;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: Joker-CC
 * @Path:
 * @Date 2021/05/10 11:51
 * @Description: RabbitMQ配置类，创建交换机，队列并绑定
 * @Version: 1.0
 */

@Configuration
@Slf4j
public class RabbitMQConfiguration {

     @Bean("defaultRabbitTemplate")
     public RabbitTemplate initRabbitTemplate(@Autowired ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
     }

    /**
     * 自定义rabbitTemplate
     * @param connectionFactory
     * 它会建立一个可由应用程序共享的连接代理。共享连接是可能的，因为使用 AMQP 进行消息传递的“工作单元”实际上是一个“通道”（在某些方面，这类似于 JMS 中连接和会话之间的关系）。
     * 连接实例提供了一种createChannel方法。该CachingConnectionFactory实现支持对这些通道进行缓存，并根据通道是否为事务性维护单独的缓存。创建 的实例时CachingConnectionFactory，您可以通过构造函数提供“主机名”。
     * 您还应该提供“用户名”和“密码”属性。要配置通道缓存的大小（默认为 25），可以调用该 setChannelCacheSize()方法
     * @return
     */
    @Bean("diyRabbitTemplate")
    public RabbitTemplate initAckRabbitTemplate(@Autowired CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
        rabbitTemplate.setMandatory(true);
        //为此模板设置消息转换器。 用于将 Object 参数解析为 convertAndSend 方法和来自 receiveAndConvert 方法的 Object 结果。默认转换器是 SimpleMessageConverter，它能够根据消息内容类型标头处理字节数组、字符串和可序列化对象。
        rabbitTemplate.setMessageConverter(converter());

        //----------------------------------------发布者重试机制-------------------------------------------------------------
        /*
        消息是否成功发送到Exchange
        参数：
            correlationData相关数据 – 回调的相关数据。
            ack – ack 为真，nack 为假
            cause 原因 – 可选原因，如果可用，则为 nack，否则为 null。
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                String msgId = correlationData.getId();
                log.info("消息成功发送到Exchange--->"+msgId);
                //msgLogService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_SUCCESS);
            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });


        /*
        消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
        参数：
            message – 返回的消息。
            回复代码 - 回复代码。
            回复文本 – 回复文本。
            交换——交换。
            routingKey – 路由密钥。
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });

        return rabbitTemplate;
    }

    public Jackson2JsonMessageConverter converter() {
        return new Jackson2JsonMessageConverter();
    }





    //1.普通队列1
    @Bean(name = RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST)
    public Queue createQueue1Test(){
        /*
        参数：
        name--->队列名称-不能为null,设置为""可以使代理生成名称。
        durable--->如果我们声明一个持久队列，则为true（该队列将在服务器重启后继续存在）
        Exclusive---> 如果我们声明一个排他队列，则为true（该队列仅由声明者的连接使用）
        autoDelete--->如果服务器在不再使用队列时应该删除队列，则为true
        arguments---> 用于声明队列的参数
         */
        return new Queue(RabbitMQBeanName.DEFAULT_QUEUE1_TO_TEST,true,false,false,null);
    }


    //2.发布订阅模式队列1
    @Bean(name = RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST)
    public Queue createPublishQueue1Test(){
        return new Queue(RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST,true,false,false,null);
    }

    //2.发布订阅模式队列2
    @Bean(name = RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST)
    public Queue createPublishQueue2Test(){
        return new Queue(RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST,true,false,false,null);
    }

    //2.发布订阅模式交换机
    @Bean(name = RabbitMQBeanName.FANOUT_EXCHANGE)
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(RabbitMQBeanName.FANOUT_EXCHANGE,true,false,null);
    }

    //2.发布订阅模式Binding1
    @Bean
    public Binding fanoutBinding1(@Qualifier(RabbitMQBeanName.FANOUT_EXCHANGE) FanoutExchange fanout, @Qualifier(RabbitMQBeanName.PUBLISH_QUEUE1_TO_TEST)Queue fanoutQueue1) {
        return BindingBuilder.bind(fanoutQueue1).to(fanout);
    }

    //2.发布订阅模式Binding2
    @Bean
    public Binding fanoutBinding2(@Qualifier(RabbitMQBeanName.FANOUT_EXCHANGE) FanoutExchange fanout, @Qualifier(RabbitMQBeanName.PUBLISH_QUEUE2_TO_TEST)Queue fanoutQueue2) {
        return BindingBuilder.bind(fanoutQueue2).to(fanout);
    }


    /**
     * 3.创建Direct交换机
     * @return
     */
    @Bean(name = RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST)
    public DirectExchange createDirectExchangeTest(){
       /*
        参数说明：
        name--->交换机的名称。
        durable--->是否持久化-如果我们声明一个持久交换机（为true该持久交换机将在服务器重新启动后继续存在）
        autoDelete--->如果服务器在不再使用该交换时应删除该交换机
        arguments--->用于声明交换的参数
         */
        return new DirectExchange(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST,true,false,null);
    }

    //3.Direct队列1
    @Bean(name = RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST)
    public Queue createDirectQueue1Test(){
        HashMap arguments = new HashMap<>(4);
        //配置延迟队列参数 标注这是一个延迟队列且消息5秒后删除，放入到死信队列
        arguments.put("x-message-ttl",5000);
        //绑定死信交换机
        arguments.put("x-dead-letter-exchange", RabbitMQBeanName.DEAD_LETTER_EXCHANGE);
        //绑定死信交换机路由key
        arguments.put("x-dead-letter-routing-key", "dead");
        //构建队列
        return new Queue(RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST,true,false,false,arguments);

    }

    //3.Direct队列1
    @Bean(name = RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST)
    public Queue createDirectQueue2Test(){
        return new Queue(RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST,true,false,false,null);
    }

    //3.路由模式Binding1
    @Bean
    public Binding directBinding1(@Qualifier(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST) DirectExchange direct, @Qualifier(RabbitMQBeanName.DIRECT_QUEUE1_TO_TEST) Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(direct).with("sb");
    }

    //3.路由模式Binding2
    @Bean
    public Binding directBinding2(@Qualifier(RabbitMQBeanName.DIRECT_EXCHANGE_TO_TEST) DirectExchange direct, @Qualifier(RabbitMQBeanName.DIRECT_QUEUE2_TO_TEST) Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(direct).with("wc");
    }




    /**
     * 4.创建topic交换机
     * @return
     */
    @Bean(name = RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST)
    public TopicExchange createTopicExchangeTest(){
       /*
        参数说明：
        name--->交换机的名称。
        durable--->是否持久化-如果我们声明一个持久交换机（为true该持久交换机将在服务器重新启动后继续存在）
        autoDelete--->如果服务器在不再使用该交换时应删除该交换机
        arguments--->用于声明交换的参数
         */
        return new TopicExchange(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST,true,false,null);
    }

    //4.topic队列1
    @Bean(name = RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST)
    public Queue createTopicQueue1Test(){
        return new Queue(RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST,true,false,false,null);
    }

    //4.topic队列2
    @Bean(name = RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST)
    public Queue createTopicQueue2Test(){
        return new Queue(RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST,true,false,false,null);
    }

    //4.主题模式Binding1
    @Bean
    public Binding topicBinding1(@Qualifier(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST) TopicExchange topic, @Qualifier(RabbitMQBeanName.TOPIC_QUEUE1_TO_TEST) Queue topicQueue1) {
        //通配符路由key
        return BindingBuilder.bind(topicQueue1).to(topic).with("#.wo.shi.sb.#");
    }

    //4.主题模式Binding2
    @Bean
    public Binding topicBinding2(@Qualifier(RabbitMQBeanName.TOPIC_EXCHANGE_TO_TEST) TopicExchange topic, @Qualifier(RabbitMQBeanName.TOPIC_QUEUE2_TO_TEST) Queue topicQueue2) {
        //通配符路由key
        return BindingBuilder.bind(topicQueue2).to(topic).with("*.wo.shi.sb.*");
    }



    //5.创建定时队列
    @Bean(name = RabbitMQBeanName.TTL_QUEUE_TO_TEST)
    public Queue createTTLQueueTest(){
        /*
        参数：
        name--->队列名称-不能为null,设置为""可以使代理生成名称。
        durable--->如果我们声明一个持久队列，则为true（该队列将在服务器重启后继续存在）
        Exclusive---> 如果我们声明一个排他队列，则为true（该队列仅由声明者的连接使用）
        autoDelete--->如果服务器在不再使用队列时应该删除队列，则为true
        arguments---> 用于声明队列的参数
         */
        HashMap arguments = new HashMap<>(4);
        //配置延迟队列参数 标注这是一个延迟队列且消息5秒后删除，放入到死信队列
        arguments.put("x-message-ttl",5000);
        //绑定死信交换机
        arguments.put("x-dead-letter-exchange", RabbitMQBeanName.DEAD_LETTER_EXCHANGE);
        //绑定死信交换机路由key
        arguments.put("x-dead-letter-routing-key", "dead");
        return new Queue(RabbitMQBeanName.TTL_QUEUE_TO_TEST,true,false,false,arguments);
    }


    //5.创建死信交换机
    @Bean(RabbitMQBeanName.DEAD_LETTER_EXCHANGE)
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(RabbitMQBeanName.DEAD_LETTER_EXCHANGE,true,false,null);
    }

    //5.创建死信队列
    @Bean(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST)
    public Queue createDeadLetterQueue1Test(){
        return new Queue(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST,true,false,false,null);
    }

    //5.死信队列和死信交换机绑定
    @Bean
    public Binding deadLetterBindingB(@Qualifier(RabbitMQBeanName.DEAD_LETTER_QUEUE_TO_TEST) Queue queue,
                                      @Qualifier(RabbitMQBeanName.DEAD_LETTER_EXCHANGE) DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("dead");
    }




}

生产者配置

spring:
  rabbitmq:
    host: xxxx
    port: xxx
    username: xxx
    password: xxx
    virtual-host: crm
    #开启发布者回调
    publisher-returns: true
    # 发送确认  具体配置可以看此类 ConfirmType
    publisher-confirm-type: correlated
    #设置手动ack
    listener:
      simple:
        #手动确认 具体配置可以看此类 AcknowledgeMode
        acknowledge-mode: MANUAL
  #    template:
  #      #当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  #      #那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;
  #      #通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者
  #      mandatory: true

消费者配置

server:
  port: 9090

spring:
  rabbitmq:
    host: xxxx
    port: xxxx
    username: xxxx
    password: xxxx
    virtual-host: scrm
    #开启发布者回调
    publisher-returns: true
    # 发送确认  具体配置可以看此类 ConfirmType
    publisher-confirm-type: correlated
    #设置手动ack
    listener:
      simple:
        #手动确认 具体配置可以看此类 AcknowledgeMode
        acknowledge-mode: MANUAL
        #每个消费者可以未确认的最大未确认消息数  具体配置可以看此类AmqpContainer具体实现类是SimpleContainer
        prefetch: 100
        #Spring官网的列子使用 retryTemplate进行配置
        retry:
          #开启发布重试 (默认false)
          enabled: true
          #传递消息的最大尝试次数 (默认3)
          max-attempts: 2
          #第一次和第二次尝试传递消息之间的持续时间  重试间隔时间（单位毫秒 默认1000ms）
          initial-interval: 5000ms
          #尝试之间的最长持续时间 （默认10000ms）
          max-interval: 20000ms
          #应用于前一个重试间隔的乘数。 （默认1.0） 间隔时间*乘数=下一次的间隔时间，最大不能超过设置的最大间隔时间
          multiplier: 1

  #    template:
  #      #当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
  #      #那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;
  #      #通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者
  #      mandatory: true