六、死信队列

6.1 死信的概念

死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

6.2 死信的来源

  1. 消息TTL过期
  2. 队列达到最大长度(队列满了，无法再添加数据到mq中)
  3. 消息被拒绝(basic.reject或basic.nack)并且requeue=false

    6.3 代码实践

    6.3.1 消息TTL过期

    ```java package com.rem.rabbitmq.ee.HdeadLetterQueue;

import com.rabbitmq.client.AMQP; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rem.rabbitmq.ee.RabbitMqUtils;

import java.util.Scanner;

/**

  • 死信队列
  • 某些时候由于特定的原因导致queue中的某些消息无法被消费 这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

  • 死信来源
    • 消息TTL过去
    • 队列达到最大长度
    • 消息被拒绝 *
  • @author Rem
  • @date 2021-12-27 */

public class TTLProducer {

  1. public static void main(String[] args) throws Exception {
  3.     Channel channel = RabbitMqUtils.getChannel();
  4.     channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_TTL, BuiltinExchangeType.DIRECT);
  6.     System.err.println("开始发送消息...");
  7.     Scanner scanner = new Scanner(System.in);
  8.     while (scanner.hasNext()) {
  9.         String message = scanner.next();
  11.         /**
  12.          * 发送消息
  13.          * 设置消息过期时间 ms
  14.          */
  15.         AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
  16.         channel.basicPublish(RabbitMqUtils.EXCHANGE_NORMAL_TTL, RabbitMqUtils.ROUTING_KEY_NORMAL_TTL, properties, message.getBytes());
  17.         System.err.println("发送消息完毕" + message);
  18.     }
  20. }

}

  1. ```java
  2. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  4. import com.rabbitmq.client.BuiltinExchangeType;
  5. import com.rabbitmq.client.CancelCallback;
  6. import com.rabbitmq.client.Channel;
  7. import com.rabbitmq.client.DeliverCallback;
  8. import com.rem.rabbitmq.ee.RabbitMqUtils;
  10. import java.util.HashMap;
  11. import java.util.Map;
  14. /**
  15.  * 普通消费者-验证ttl过期
  16.  *
  17.  * @author Rem
  18.  * @date 2021-12-27
  19.  */
  21. public class ConsumerTTL15 {
  23.     public static void main(String[] args) throws Exception {
  25.         Channel channel = RabbitMqUtils.getChannel();
  27.         //声明死信交换机
  28.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DEAD, BuiltinExchangeType.DIRECT);
  29.         //声明死信队列
  30.         channel.queueDeclare(RabbitMqUtils.QUEUE_DEAD, false, false, false, null);
  31.         //死信队列绑定死信交换机与routingkey
  32.         channel.queueBind(RabbitMqUtils.QUEUE_DEAD, RabbitMqUtils.EXCHANGE_DEAD, RabbitMqUtils.ROUTING_KEY_DEAD);
  34.         /********************************************************************************/
  36.         //声明普通交换机
  37.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_TTL, BuiltinExchangeType.DIRECT);
  38.         //声明普通队列 并且绑定死信交换机
  39.         //设置参数
  40.         Map<String, Object> arguments = new HashMap<>();
  41.         //设置过期时间 单位ms  10s  一般都在生产方设置
  42.         //arguments.put("x-message-ttl", 10000);
  43.         arguments.put("x-dead-letter-exchange", RabbitMqUtils.EXCHANGE_DEAD);
  44.         arguments.put("x-dead-letter-routing-key", RabbitMqUtils.ROUTING_KEY_DEAD);
  45.         channel.queueDeclare(RabbitMqUtils.QUEUE_NORMAL_TTL, false, false, false, arguments);
  47.         //普通交换机与普通队列通过普通routingkey 进行绑定
  48.         channel.queueBind(RabbitMqUtils.QUEUE_NORMAL_TTL, RabbitMqUtils.EXCHANGE_NORMAL_TTL, RabbitMqUtils.ROUTING_KEY_NORMAL_TTL);
  50.         //消费回调
  51.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  53.         //取消消费回调 如果在消费的时候队列被删除了
  54.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  56.         System.out.println("消费者-NORMAL-TTL 等待消费");
  57.         channel.basicConsume(RabbitMqUtils.QUEUE_NORMAL_TTL, true, deliverCallback, cancelCallback);
  58.     }
  61. }
  1. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * 死信消费者
  11.  *
  12.  * @author Rem
  13.  * @date 2021-12-27
  14.  */
  16. public class ConsumerDead16 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  22.         //声明死信交换机
  23.        // channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DEAD, BuiltinExchangeType.DIRECT);
  24.         //声明死信队列
  25.        // channel.queueDeclare(RabbitMqUtils.QUEUE_DEAD, false, false, false, null);
  26.         //死信队列绑定死信交换机与routingkey
  27.        // channel.queueBind(RabbitMqUtils.QUEUE_DEAD, RabbitMqUtils.EXCHANGE_DEAD, RabbitMqUtils.ROUTING_KEY_DEAD);
  29.         //消费回调
  30.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的死信消息：" + new String(message.getBody()));
  32.         //取消消费回调 如果在消费的时候队列被删除了
  33.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  35.         System.out.println("消费者-DEAD 等待消费");
  36.         channel.basicConsume(RabbitMqUtils.QUEUE_DEAD, true, deliverCallback, cancelCallback);
  37.     }
  40. }

6.3.2 消息到达最大长度

  1. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.Scanner;
  9. /**
  10.  * 死信队列
  11.  * 某些时候由于特定的原因导致queue中的某些消息无法被消费 这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
  12.  * <p>
  13.  * 死信来源
  14.  * * 消息TTL过去
  15.  * * 队列达到最大长度
  16.  * * 消息被拒绝
  17.  *
  18.  * @author Rem
  19.  * @date 2021-12-29
  20.  */
  22. public class MaxLengthProducer {
  25.     public static void main(String[] args) throws Exception {
  27.         Channel channel = RabbitMqUtils.getChannel();
  28.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_MAXLENGTH, BuiltinExchangeType.DIRECT);
  30.         System.err.println("开始发送消息...");
  31.         Scanner scanner = new Scanner(System.in);
  32.         while (scanner.hasNext()) {
  33.             String message = scanner.next();
  35.             channel.basicPublish(RabbitMqUtils.EXCHANGE_NORMAL_MAXLENGTH, RabbitMqUtils.ROUTING_KEY_NORMAL_MAXLENGTH, null, message.getBytes());
  36.             System.err.println("发送消息完毕" + message);
  37.         }
  39.     }
  42. }
  1. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.CancelCallback;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.DeliverCallback;
  7. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. import java.util.HashMap;
  10. import java.util.Map;
  13. /**
  14.  * 普通消费者-验证队列最大长度
  15.  *
  16.  * @author Rem
  17.  * @date 2021-12-27
  18.  */
  20. public class ConsumerMaxLength17 {
  22.     public static void main(String[] args) throws Exception {
  24.         Channel channel = RabbitMqUtils.getChannel();
  26.         //声明死信交换机
  27.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DEAD, BuiltinExchangeType.DIRECT);
  28.         //声明死信队列
  29.         channel.queueDeclare(RabbitMqUtils.QUEUE_DEAD, false, false, false, null);
  30.         //死信队列绑定死信交换机与routingkey
  31.         channel.queueBind(RabbitMqUtils.QUEUE_DEAD, RabbitMqUtils.EXCHANGE_DEAD, RabbitMqUtils.ROUTING_KEY_DEAD);
  33.         /********************************************************************************/
  35.         //声明普通交换机
  36.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_MAXLENGTH, BuiltinExchangeType.DIRECT);
  37.         //声明普通队列 并且绑定死信交换机
  38.         //设置参数
  39.         Map<String, Object> arguments = new HashMap<>();
  40.         arguments.put("x-dead-letter-exchange", RabbitMqUtils.EXCHANGE_DEAD);
  41.         arguments.put("x-dead-letter-routing-key", RabbitMqUtils.ROUTING_KEY_DEAD);
  42.         /**
  43.          * 设置队列最大长度
  44.          */
  45.         arguments.put("x-max-length", 6);
  46.         channel.queueDeclare(RabbitMqUtils.QUEUE_NORMAL_MAXLENGTH, false, false, false, arguments);
  48.         //普通交换机与普通队列通过普通routingkey 进行绑定
  49.         channel.queueBind(RabbitMqUtils.QUEUE_NORMAL_MAXLENGTH, RabbitMqUtils.EXCHANGE_NORMAL_MAXLENGTH, RabbitMqUtils.ROUTING_KEY_NORMAL_MAXLENGTH);
  51.         //消费回调
  52.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  54.         //取消消费回调 如果在消费的时候队列被删除了
  55.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  57.         System.out.println("消费者-NORMAL-MaxLength 等待消费");
  58.         channel.basicConsume(RabbitMqUtils.QUEUE_NORMAL_MAXLENGTH, true, deliverCallback, cancelCallback);
  59.     }
  62. }

6.3.3 消息被拒绝

  1. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.Scanner;
  9. /**
  10.  * 死信队列
  11.  * 某些时候由于特定的原因导致queue中的某些消息无法被消费 这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
  12.  * <p>
  13.  * 死信来源
  14.  * * 消息TTL过去
  15.  * * 队列达到最大长度
  16.  * * 消息被拒绝
  17.  *
  18.  * @author Rem
  19.  * @date 2021-12-29
  20.  */
  22. public class RejectProducer {
  25.     public static void main(String[] args) throws Exception {
  27.         Channel channel = RabbitMqUtils.getChannel();
  28.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_REJECT, BuiltinExchangeType.DIRECT);
  30.         System.err.println("开始发送消息...");
  31.         Scanner scanner = new Scanner(System.in);
  32.         while (scanner.hasNext()) {
  33.             String message = scanner.next();
  35.             channel.basicPublish(RabbitMqUtils.EXCHANGE_NORMAL_REJECT, RabbitMqUtils.ROUTING_KEY_NORMAL_REJECT, null, message.getBytes());
  36.             System.err.println("发送消息完毕" + message);
  37.         }
  39.     }
  42. }
  1. package com.rem.rabbitmq.ee.HdeadLetterQueue;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.CancelCallback;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.DeliverCallback;
  7. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. import java.util.HashMap;
  10. import java.util.Map;
  13. /**
  14.  * 普通消费者-验证拒绝消息
  15.  *
  16.  * @author Rem
  17.  * @date 2021-12-27
  18.  */
  20. public class ConsumerReject18 {
  22.     public static void main(String[] args) throws Exception {
  24.         Channel channel = RabbitMqUtils.getChannel();
  26.         //声明死信交换机
  27.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DEAD, BuiltinExchangeType.DIRECT);
  28.         //声明死信队列
  29.         channel.queueDeclare(RabbitMqUtils.QUEUE_DEAD, false, false, false, null);
  30.         //死信队列绑定死信交换机与routingkey
  31.         channel.queueBind(RabbitMqUtils.QUEUE_DEAD, RabbitMqUtils.EXCHANGE_DEAD, RabbitMqUtils.ROUTING_KEY_DEAD);
  33.         /********************************************************************************/
  35.         //声明普通交换机
  36.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_REJECT, BuiltinExchangeType.DIRECT);
  37.         //声明普通队列 并且绑定死信交换机
  38.         //设置参数
  39.         Map<String, Object> arguments = new HashMap<>();
  40.         arguments.put("x-dead-letter-exchange", RabbitMqUtils.EXCHANGE_DEAD);
  41.         arguments.put("x-dead-letter-routing-key", RabbitMqUtils.ROUTING_KEY_DEAD);
  42.         /**
  43.          * 设置队列最大长度
  44.          */
  45.         arguments.put("x-max-length", 6);
  46.         channel.queueDeclare(RabbitMqUtils.QUEUE_NORMAL_REJECT, false, false, false, arguments);
  48.         //普通交换机与普通队列通过普通routingkey 进行绑定
  49.         channel.queueBind(RabbitMqUtils.QUEUE_NORMAL_REJECT, RabbitMqUtils.EXCHANGE_NORMAL_REJECT, RabbitMqUtils.ROUTING_KEY_NORMAL_REJECT);
  51.         //消费回调
  52.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  53.             System.out.println("接收到的消息,并且拒绝：" + new String(message.getBody()));
  54.             /**
  55.              *  拒绝消息
  56.              *  DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
  57.              *  requeue – 如果被拒绝的消息应该重新排队而不是丢弃/死信，则为真
  58.              */
  59.             channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
  60.         };
  62.         //取消消费回调 如果在消费的时候队列被删除了
  63.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  65.         System.out.println("消费者-NORMAL-REJECT 等待消费");
  67.         /**
  68.          * 设置自动应答为false
  69.          */
  70.         boolean autoAck = false;
  71.         channel.basicConsume(RabbitMqUtils.QUEUE_NORMAL_REJECT, autoAck, deliverCallback, cancelCallback);
  72.     }
  75. }

七、延迟队列

使用死信队列实现延迟队列

  1. package com.rem.rabbitmq.ee.Idelay;
  3. import com.rabbitmq.client.AMQP;
  4. import com.rabbitmq.client.BuiltinExchangeType;
  5. import com.rabbitmq.client.Channel;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  8. import java.util.Scanner;
  10. /**
  11.  * 延迟队列 -死信队列中ttl的实现
  12.  * 设置ttl过期两种方法
  13.  * ** 在队列上设置过期时间  arguments.put("x-message-ttl", 10000)  一旦消息过期就被丢弃，如果配置了死信队列就丢到死信队列
  14.  * ** 在消息上设置过期时间  new AMQP.BasicProperties().builder().expiration("10000").build()  如果不设置代表永不过期；如果设置为0代表直接发送给消费者，否则丢弃消息
  15.  *
  16.  * @author Rem
  17.  * @date 2021-12-27
  18.  */
  20. public class DelayProducer {
  23.     public static void main(String[] args) throws Exception {
  25.         Channel channel = RabbitMqUtils.getChannel();
  26.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_TTL, BuiltinExchangeType.DIRECT);
  28.         System.err.println("开始发送消息...");
  29.         Scanner scanner = new Scanner(System.in);
  30.         while (scanner.hasNext()) {
  31.             String message = scanner.next();
  33.             /**
  34.              * 发送消息
  35.              * 设置消息过期时间 ms
  36.              */
  37.             AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("8000").build();
  38.             channel.basicPublish(RabbitMqUtils.EXCHANGE_NORMAL_TTL, RabbitMqUtils.ROUTING_KEY_NORMAL_TTL, properties, message.getBytes());
  39.             System.err.println("发送消息完毕" + message);
  40.         }
  42.     }
  45. }
  1. package com.rem.rabbitmq.ee.Idelay;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.CancelCallback;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.DeliverCallback;
  7. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. import java.util.HashMap;
  10. import java.util.Map;
  13. /**
  14.  * 普通消费者-延迟队列
  15.  *
  16.  * @author Rem
  17.  * @date 2021-12-27
  18.  */
  20. public class Consumer19 {
  22.     public static void main(String[] args) throws Exception {
  24.         Channel channel = RabbitMqUtils.getChannel();
  26.         //声明死信交换机
  27.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DEAD, BuiltinExchangeType.DIRECT);
  28.         //声明死信队列
  29.         channel.queueDeclare(RabbitMqUtils.QUEUE_DEAD, false, false, false, null);
  30.         //死信队列绑定死信交换机与routingkey
  31.         channel.queueBind(RabbitMqUtils.QUEUE_DEAD, RabbitMqUtils.EXCHANGE_DEAD, RabbitMqUtils.ROUTING_KEY_DEAD);
  33.         /********************************************************************************/
  35.         //声明普通交换机
  36.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_NORMAL_TTL, BuiltinExchangeType.DIRECT);
  37.         //声明普通队列 并且绑定死信交换机
  38.         //设置参数
  39.         Map<String, Object> arguments = new HashMap<>();
  40.         //设置过期时间 单位ms  10s  一般都在生产方设置
  41.         //arguments.put("x-message-ttl", 10000);
  42.         arguments.put("x-dead-letter-exchange", RabbitMqUtils.EXCHANGE_DEAD);
  43.         arguments.put("x-dead-letter-routing-key", RabbitMqUtils.ROUTING_KEY_DEAD);
  44.         channel.queueDeclare(RabbitMqUtils.QUEUE_NORMAL_TTL, false, false, false, arguments);
  46.         //消费回调
  47.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的死信消息：" + new String(message.getBody()));
  49.         //取消消费回调 如果在消费的时候队列被删除了
  50.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  52.         /**
  53.          * 直接延迟队列接收消息
  54.          */
  55.         System.out.println("消费者-DEAD 等待消费");
  56.         channel.basicConsume(RabbitMqUtils.QUEUE_DEAD, true, deliverCallback, cancelCallback);
  57.     }
  60. }

安装插件实现延迟队列

rabbitmq 插件安装
延迟队列插件下载

  1. #拷贝到rabbitmq容器 773067241f96 中
  3. docker cp /home/rabbitmq/rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins
  5. #进入容器
  7. docker exec -it rabbitmq /bin/bash
  9. #启用插件
  11. rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  13. #查看
  15. rabbitmq-plugins list
  17. #重新启动容器
  19. docker restart rabbitmq

集成springboot代码实现

  1. server:
  2.   servlet:
  3.     context-path: /rabbit
  4. spring:
  5.   rabbitmq:
  6.     host: remzhi.top
  7.     port: 5672
  8.     username: rem
  9.     password: 123456
  1. package com.rem.rabbitmq.boot.Adelay;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.DirectExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.amqp.core.QueueBuilder;
  8. import org.springframework.beans.factory.annotation.Qualifier;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.context.annotation.Configuration;
  12. /**
  13.  * 使用死信队列实现 延迟队列
  14.  *
  15.  * @author Rem
  16.  * @date 2021-12-29
  17.  */
  19. @Configuration
  20. public class QueueConfig {
  22.     /**
  23.      * 普通交换机 队列 routingKey
  24.      */
  25.     public static final String X_EXCHANGE = "X";
  27.     public static final String XA_ROUTING_KEY = "XA";
  29.     public static final String XB_ROUTING_KEY = "XB";
  31.     public static final String QA_QUEUE = "QA";
  33.     public static final String QB_QUEUE = "QB";
  35.     public static final String XC_ROUTING_KEY = "XC";
  37.     public static final String QC_QUEUE = "QC";
  40.     /**
  41.      * 死信交换机 队列 routingKey
  42.      */
  43.     public static final String Y_DEAD_EXCHANGE = "Y";
  45.     public static final String YD_DEAD_ROUTING_KEY = "YD";
  47.     public static final String QD_DEAD_QUEUE = "QD";
  50.     /*********************************普通队列QA*********************************************************/
  51.     /**
  52.      * 声明普通交换机X
  53.      *
  54.      * @return
  55.      */
  56.     @Bean("xExchange")
  57.     public DirectExchange xExchange() {
  58.         return new DirectExchange(X_EXCHANGE);
  59.     }
  61.     /**
  62.      * 声明普通队列QA并且绑定死信交换机
  63.      *
  64.      * @return
  65.      */
  66.     @Bean("queueA")
  67.     public Queue queueA() {
  68.         /**
  69.          * 手动设置参数
  70.          */
  71.        /* Map<String, Object> arguments = new HashMap<>();
  72.         arguments.put("x-dead-letter-exchange", Y_DEAD_EXCHANGE);
  73.         arguments.put("x-dead-letter-routing-key", YD_DEAD_ROUTING_KEY);
  74.         arguments.put("x-message-ttl", 10000);
  75.         return QueueBuilder.durable(QA_QUEUE).withArguments(arguments).build();*/
  76.         /**
  77.          * 根据内置api设置参数 过期时间 ms
  78.          */
  79.         return QueueBuilder.durable(QA_QUEUE)
  80.                 .deadLetterExchange(Y_DEAD_EXCHANGE).deadLetterRoutingKey(YD_DEAD_ROUTING_KEY).ttl(10000).build();
  81.     }
  83.     /**
  84.      * 普通交换机与普通队列QA通过 普通routingKey XA 绑定
  85.      *
  86.      * @param queueA
  87.      * @param xExchange
  88.      * @return
  89.      */
  90.     @Bean
  91.     public Binding queueaBindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange) {
  92.         return BindingBuilder.bind(queueA).to(xExchange).with(XA_ROUTING_KEY);
  93.     }
  95.     /*********************************普通队列QB*********************************************************/
  97.     /**
  98.      * 声明普通队列QB 并且绑定死信交换机
  99.      *
  100.      * @return
  101.      */
  102.     @Bean("queueB")
  103.     public Queue queueB() {
  104.         /**
  105.          * 根据内置api设置参数
  106.          */
  107.         return QueueBuilder.durable(QB_QUEUE)
  108.                 .deadLetterExchange(Y_DEAD_EXCHANGE).deadLetterRoutingKey(YD_DEAD_ROUTING_KEY).ttl(40000).build();
  109.     }
  111.     /**
  112.      * 普通交换机与普通队列通过QB 普通routingKey 绑定
  113.      *
  114.      * @param queueB
  115.      * @param xExchange
  116.      * @return
  117.      */
  118.     @Bean
  119.     public Binding queuebBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
  120.         return BindingBuilder.bind(queueB).to(xExchange).with(XB_ROUTING_KEY);
  121.     }
  124.     /*********************************死信队列QD*********************************************************/
  127.     /**
  128.      * 声明死信交换机Y
  129.      *
  130.      * @return
  131.      */
  132.     @Bean("yExchange")
  133.     public DirectExchange yExchange() {
  134.         return new DirectExchange(Y_DEAD_EXCHANGE);
  135.     }
  137.     /**
  138.      * 声明死信队列
  139.      *
  140.      * @return
  141.      */
  142.     @Bean("queueD")
  143.     public Queue queueD() {
  144.         return new Queue(QD_DEAD_QUEUE);
  145.     }
  147.     /**
  148.      * 死信交换机与死信队列通过 死信routingKey XA 绑定
  149.      *
  150.      * @param queueD
  151.      * @param yExchange
  152.      * @return
  153.      */
  154.     @Bean
  155.     public Binding deadLetterBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange) {
  156.         return BindingBuilder.bind(queueD).to(yExchange).with(YD_DEAD_ROUTING_KEY);
  157.     }
  160.     /*********************************普通队列QC*********************************************************/
  162.     /**
  163.      * 声明普通队列QC 并且绑定死信交换机
  164.      * 不设置过期时间 交给生产方设置
  165.      * *因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，
  166.      * *而第二个消息的延时时长很短，第二个消息并不会优先得到执行。
  167.      *
  168.      * @return
  169.      */
  170.     @Bean("queueC")
  171.     public Queue queueC() {
  172.         /**
  173.          * 根据内置api设置参数
  174.          */
  175.         return QueueBuilder.durable(QC_QUEUE)
  176.                 .deadLetterExchange(Y_DEAD_EXCHANGE).deadLetterRoutingKey(YD_DEAD_ROUTING_KEY).build();
  177.     }
  179.     /**
  180.      * 普通交换机与普通队列通过QC 普通routingKey 绑定
  181.      *
  182.      * @param queueC
  183.      * @param xExchange
  184.      * @return
  185.      */
  186.     @Bean
  187.     public Binding queuecBindingX(@Qualifier("queueC") Queue queueC, @Qualifier("xExchange") DirectExchange xExchange) {
  188.         return BindingBuilder.bind(queueC).to(xExchange).with(XC_ROUTING_KEY);
  189.     }
  192. }
  1. package com.rem.rabbitmq.boot.Adelay;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * 使用插件实现rabbitMQ 延迟队列
  10.  *
  11.  * @author Rem
  12.  * @date 2022-01-03
  13.  */
  15. @Configuration
  16. public class DelayQueueConfig {
  18.     /**
  19.      * 延迟交换机 队列 routingKey
  20.      */
  21.     public static final String DELAY_EXCHANGE = "DELAY_EXCHANGE";
  23.     public static final String DELAY_ROUTING_KEY = "DELAY_ROUTING_KEY";
  25.     public static final String DELAY_QUEUE = "DELAY_QUEUE";
  27.     @Bean
  28.     public DirectExchange delayExchange() {
  29.         //自定义参数
  30. //        Map<String, Object> arguments = new HashMap<>();
  31. //        //自定义交换机的类型
  32. //        arguments.put("x-delayed-type", "direct");
  33. //        return new CustomExchange(DELAY_EXCHANGE, "x-delayed-message", true, false, arguments);
  35.         //使用api设置
  36.         return ExchangeBuilder.directExchange(DELAY_EXCHANGE).delayed().build();
  37.     }
  39.     @Bean
  40.     public Queue delayQueue() {
  41.         return QueueBuilder.durable(DELAY_QUEUE).build();
  42.     }
  44.     @Bean
  45.     public Binding bindingDelayedQueue(@Qualifier("delayQueue") Queue delayQueue, @Qualifier("delayExchange") Exchange delayExchange) {
  46.         return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
  47.     }
  49. }
  1. package com.rem.rabbitmq.boot.Adelay;
  3. import io.swagger.annotations.Api;
  4. import io.swagger.annotations.ApiOperation;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.core.MessagePostProcessor;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.web.bind.annotation.GetMapping;
  10. import org.springframework.web.bind.annotation.PathVariable;
  11. import org.springframework.web.bind.annotation.RequestMapping;
  12. import org.springframework.web.bind.annotation.RestController;
  14. import java.time.LocalDateTime;
  16. /**
  17.  * @author Rem
  18.  * @date 2021-12-30
  19.  */
  21. @Slf4j
  22. @RestController
  23. @RequestMapping("/delay")
  24. @Api(tags = "延迟队列")
  25. public class SendMessageController {
  27.     @Autowired
  28.     private RabbitTemplate rabbitTemplate;
  31.     @ApiOperation(value = "队列设置过期时间 的延迟消息")
  32.     @GetMapping("/sendMsg/{message}")
  33.     public void sendMsg(@PathVariable String message) {
  35.         log.info("当前时间：{},发送一条信息给两个TTL队列:{}", LocalDateTime.now(), message);
  37.         rabbitTemplate.convertAndSend(QueueConfig.X_EXCHANGE, QueueConfig.XA_ROUTING_KEY, "消息来自ttl为10S的队列:" + message);
  39.         rabbitTemplate.convertAndSend(QueueConfig.X_EXCHANGE, QueueConfig.XB_ROUTING_KEY, "消息来自ttl为40S的队列:" + message);
  40.     }
  42.     @ApiOperation(value = "消息设置过期时间 的延迟消息")
  43.     @GetMapping("/sendMsg/{message}/{ttlTime}")
  44.     public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
  45.         log.info("当前时间：{},发送一条信息:{}，,并且设置ttl:{}", LocalDateTime.now(), message, ttlTime);
  47.         /**
  48.          * setExpiration 设置过期时间
  49.          */
  50.         MessagePostProcessor messagePostProcessor = messageProperties -> {
  51.             messageProperties.getMessageProperties().setExpiration(ttlTime);
  52.             return messageProperties;
  53.         };
  54.         rabbitTemplate.convertAndSend(QueueConfig.X_EXCHANGE, QueueConfig.XC_ROUTING_KEY, message, messagePostProcessor);
  55.     }
  57.     @ApiOperation(value = "设置插件-发送生产端设置过期时间 的延迟消息")
  58.     @GetMapping("/sendDelayMsg/{message}/{ttlTime}")
  59.     public void sendDelayMsg(@PathVariable String message, @PathVariable Integer ttlTime) {
  60.         log.info("当前时间：{},发送一条信息:{}，,使用延迟队列插件 并且设置ttl:{}", LocalDateTime.now(), message, ttlTime);
  62.         /**
  63.          * setDelay  设置延迟时间
  64.          */
  65.         MessagePostProcessor messagePostProcessor = messageProperties -> {
  66.             messageProperties.getMessageProperties().setDelay(ttlTime);
  67.             return messageProperties;
  68.         };
  69.         rabbitTemplate.convertAndSend(DelayQueueConfig.DELAY_EXCHANGE, DelayQueueConfig.DELAY_ROUTING_KEY, message, messagePostProcessor);
  70.     }
  72. }
  2.     /**
  3.      * 监听死信队列里的消息
  4.      *
  5.      * @param message
  6.      */
  7.     @RabbitListener(queues = QueueConfig.QD_DEAD_QUEUE)
  8.     public void receiveD(Message message) {
  9.         String msg = new String(message.getBody());
  10.         log.info("当前时间：{},收到死信队列信息{}", LocalDateTime.now(), msg);
  12.     }
  14.     /**
  15.      * 监听使用插件的延迟队列消息
  16.      *
  17.      * @param message
  18.      */
  19.     @RabbitListener(queues = DelayQueueConfig.DELAY_QUEUE)
  20.     public void receiveDelay(Message message) {
  21.         String msg = new String(message.getBody());
  22.         log.info("当前时间：{},收到延迟队列信息{}", LocalDateTime.now(), msg);
  24.     }

八、发布确认高级

  1. server:
  2.   servlet:
  3.     context-path: /rabbit
  4. spring:
  5.   rabbitmq:
  6.     host: remzhi.top
  7.     port: 5672
  8.     username: rem
  9.     password: 123456
  10.     #设置发布确认模式
  11.     # correlated ：发布消息成功到交换器后会触发回调方法
  12.     # simple:其一效果和CORRELATED值一样会触发回调方法
  13.     #其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法 等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑
  14.     #要注意的点是 waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker
  15.     publisher-confirm-type: correlated
  16.     #设置消息回退 默认false
  17.     #true时 当无法被路由的消息会回退给生产者
  18.     publisher-returns: true
  1. package com.rem.rabbitmq.boot.Bconfirm;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * @author Rem
  10.  * @date 2021-12-29
  11.  */
  13. @Configuration
  14. public class ConfirmConfig {
  16.     /**
  17.      * 发布确认交换机 队列 routingKey
  18.      */
  19.     public static final String CONFIRM_EXCHANGE = "CONFIRM_EXCHANGE";
  21.     public static final String CONFIRM_ROUTING_KEY = "CONFIRM_ROUTING_KEY";
  23.     public static final String CONFIRM_QUEUE = "CONFIRM_QUEUE";
  25.     /**
  26.      * 声明业务交换机
  27.      *
  28.      * @return
  29.      */
  30.     @Bean("confirmExchange")
  31.     public DirectExchange confirmExchange() {
  32.         return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).build();
  33.     }
  35.     /**
  36.      * 声明发布确认队列
  37.      *
  38.      * @return
  39.      */
  40.     @Bean("confirmQueue")
  41.     public Queue confirmQueue() {
  42.         return QueueBuilder.durable(CONFIRM_QUEUE).build();
  43.     }
  45.     /**
  46.      * 声明绑定关系
  47.      *
  48.      * @param confirmQueue
  49.      * @param confirmExchange
  50.      * @return
  51.      */
  52.     @Bean
  53.     public Binding confirmQueueaBinding(@Qualifier("confirmQueue") Queue confirmQueue, @Qualifier("confirmExchange") DirectExchange confirmExchange) {
  54.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
  55.     }
  58. }
  1. package com.rem.rabbitmq.boot.Bconfirm;
  3. import io.swagger.annotations.Api;
  4. import io.swagger.annotations.ApiOperation;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.rabbit.connection.CorrelationData;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.web.bind.annotation.GetMapping;
  10. import org.springframework.web.bind.annotation.PathVariable;
  11. import org.springframework.web.bind.annotation.RequestMapping;
  12. import org.springframework.web.bind.annotation.RestController;
  14. import java.time.LocalDateTime;
  16. /**
  17.  * @author Rem
  18.  * @date 2021-12-30
  19.  */
  21. @Slf4j
  22. @RestController
  23. @RequestMapping("/comfirm")
  24. @Api(tags = "发布确认")
  25. public class ConfirmController {
  27.     @Autowired
  28.     private RabbitTemplate rabbitTemplate;
  30.     @ApiOperation(value = "发送一条确认的消息")
  31.     @GetMapping("/sendMsg/{message}")
  32.     public void sendMsg(@PathVariable String message) {
  33.         log.info("当前时间：{}，发送消息开始:{}", LocalDateTime.now(), message);
  35.         CorrelationData correlationData = new CorrelationData("111");
  36.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUTING_KEY, ConfirmConfig.CONFIRM_ROUTING_KEY + ":" + message, correlationData);
  38.         /**
  39.          * 发送到一个假的exchange上
  40.          * 消息ack为false
  41.          */
  42.         CorrelationData correlationData2 = new CorrelationData("222");
  44.         rabbitTemplate.convertAndSend("FAKE_EXCHANGE", ConfirmConfig.CONFIRM_ROUTING_KEY, ConfirmConfig.CONFIRM_ROUTING_KEY + ":" + message, correlationData2);
  46.         /**
  47.          * 发送到一个假的routingKey上
  48.          * 消息会被退回
  49.          */
  50.         CorrelationData correlationData3 = new CorrelationData("333");
  51.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, "FAKE_ROUTING_KEY", "FAKE_ROUTING_KEY" + ":" + message, correlationData3);
  54.         log.info("当前时间：{}，发送消息完毕:{}", LocalDateTime.now(), message);
  55.     }
  58. }
  1. /**
  2.      * 发布确认队列监听
  3.      *
  4.      * @param message
  5.      */
  6.     @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
  7.     public void receiveConfirem(Message message) {
  8.         String msg = new String(message.getBody());
  9.         log.info("发布确认消费者收到的消息是:{}", msg);
  10.     }
  12.     /**
  13.      * 报警队列监听
  14.      *
  15.      * @param message
  16.      */
  17.     @RabbitListener(queues = BackupExchangeConfig.WARING_QUEUE)
  18.     public void receiveBackupWaring(Message message) {
  19.         String msg = new String(message.getBody());
  20.         log.info("报警队列收到的消息是:{}", msg);
  21.     }

备份交换机

  1. package com.rem.rabbitmq.boot.CbackupExchange;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * @author Rem
  10.  * @date 2022/1/8
  11.  */
  13. @Configuration
  14. public class BackupExchangeConfig {
  16.     /**
  17.      * 普通交换机（发布确认） 队列 routingKey
  18.      */
  19.     public static final String NORMAL_EXCHANGE = "NORMAL_EXCHANGE";
  21.     public static final String NORMAL_ROUTING_KEY = "NORMAL_ROUTING_KEY";
  23.     public static final String NORMAL_QUEUE = "NORMAL_QUEUE";
  25.     /**
  26.      * 备份交换机 备份队列 报警队列
  27.      */
  28.     public static final String BACKUP_EXCHANGE = "BACKUP_EXCHANGE";
  30.     public static final String BACKUP_QUEUE = "BACKUP_QUEUE";
  32.     public static final String WARING_QUEUE = "WARING_QUEUE";
  35.     /**
  36.      * 声明备份交换机
  37.      *
  38.      * @return
  39.      */
  40.     @Bean("backupExchange")
  41.     public FanoutExchange backupExchange() {
  42.         return ExchangeBuilder.fanoutExchange(BACKUP_EXCHANGE).build();
  43.     }
  45.     /**
  46.      * 声明备份队列
  47.      *
  48.      * @return
  49.      */
  50.     @Bean("backupQueue")
  51.     public Queue backupQueue() {
  52.         return QueueBuilder.durable(BACKUP_QUEUE).build();
  53.     }
  55.     /**
  56.      * 声明报警队列
  57.      *
  58.      * @return
  59.      */
  60.     @Bean("waringQueue")
  61.     public Queue waringQueue() {
  62.         return QueueBuilder.durable(WARING_QUEUE).build();
  63.     }
  65.     /**
  66.      * 声明备份队列绑定关系
  67.      *
  68.      * @param backupQueue
  69.      * @param backupExchange
  70.      * @return
  71.      */
  72.     @Bean
  73.     public Binding backupQueueBinding(@Qualifier("backupQueue") Queue backupQueue, @Qualifier("backupExchange") FanoutExchange backupExchange) {
  74.         return BindingBuilder.bind(backupQueue).to(backupExchange);
  75.     }
  77.     /**
  78.      * 声明报警队列绑定关系
  79.      *
  80.      * @param waringQueue
  81.      * @param backupExchange
  82.      * @return
  83.      */
  84.     @Bean
  85.     public Binding waringQueueaBinding(@Qualifier("waringQueue") Queue waringQueue, @Qualifier("backupExchange") FanoutExchange backupExchange) {
  86.         return BindingBuilder.bind(waringQueue).to(backupExchange);
  87.     }
  90.     /**
  91.      * 声明业务交换机
  92.      * **绑定备份交换机
  93.      *
  94.      * @return
  95.      */
  96.     @Bean("normalExchange")
  97.     public DirectExchange normalExchange() {
  98.         return ExchangeBuilder.directExchange(NORMAL_EXCHANGE).alternate(BACKUP_EXCHANGE).build();
  99.     }
  101.     /**
  102.      * 声明发布确认队列
  103.      *
  104.      * @return
  105.      */
  106.     @Bean("normalQueue")
  107.     public Queue normalQueue() {
  108.         return QueueBuilder.durable(NORMAL_QUEUE).build();
  109.     }
  111.     /**
  112.      * 声明普通绑定关系
  113.      *
  114.      * @param normalQueue
  115.      * @param normalExchange
  116.      * @return
  117.      */
  118.     @Bean
  119.     public Binding normalQueueaBinding(@Qualifier("normalQueue") Queue normalQueue, @Qualifier("normalExchange") DirectExchange normalExchange) {
  120.         return BindingBuilder.bind(normalQueue).to(normalExchange).with(NORMAL_ROUTING_KEY);
  121.     }
  124. }
  1. package com.rem.rabbitmq.boot;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.core.ReturnedMessage;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  10. import javax.annotation.PostConstruct;
  12. /**
  13.  * @author Rem
  14.  * @date 2022/1/5
  15.  */
  16. @Component
  17. @Slf4j
  18. public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
  20.     @Autowired
  21.     private RabbitTemplate rabbitTemplate;
  23.     //依赖注入rabbitTemplate之后再设置它的回调对象
  24.     @PostConstruct
  25.     public void init() {
  26.         rabbitTemplate.setConfirmCallback(this);
  27.         rabbitTemplate.setReturnsCallback(this);
  28.     }
  30.     /**
  31.      * 回调接口
  32.      * 当发送的消息传给不存在的交换机 ack为false
  33.      * 当发送的消息传给存在的交换机，但是routingkey错误时 ack依然是true 表示交换机收到了消息但是是路由出错
  34.      *
  35.      * @param correlationData 发布确认基类
  36.      * @param ack
  37.      * @param cause
  38.      */
  39.     @Override
  40.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  41.         if (correlationData != null) {
  42.             String id = correlationData.getId();
  43.             if (ack) {
  44.                 log.info("交换机已经收到id为:{}的消息", id);
  45.             } else {
  46.                 log.warn("交换机还未收到id为:{}消息,由于原因:{}", id, cause);
  47.             }
  48.         }
  50.     }
  52.     /**
  53.      * 只有消息不可到达时 才会回退消息
  54.      * 回退消息
  55.      *
  56.      * @param returned
  57.      */
  58.     @Override
  59.     public void returnedMessage(ReturnedMessage returned) {
  60.         String message = new String(returned.getMessage().getBody());
  61.         log.error("消息{},被交换机{}退回，退回原因:{},路由key:{}",
  62.                 message, returned.getExchange(), returned.getReplyText(), returned.getRoutingKey());
  63.     }
  64. }
  1. package com.rem.rabbitmq.boot.CbackupExchange;
  3. import io.swagger.annotations.Api;
  4. import io.swagger.annotations.ApiOperation;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.rabbit.connection.CorrelationData;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.web.bind.annotation.GetMapping;
  10. import org.springframework.web.bind.annotation.PathVariable;
  11. import org.springframework.web.bind.annotation.RequestMapping;
  12. import org.springframework.web.bind.annotation.RestController;
  14. import java.time.LocalDateTime;
  16. /**
  17.  * @author Rem
  18.  * @date 2022/1/8
  19.  */
  21. @Slf4j
  22. @RestController
  23. @RequestMapping("/backup")
  24. @Api(tags = "备份交换机")
  25. public class BackupExchangeController {
  27.     @Autowired
  28.     private RabbitTemplate rabbitTemplate;
  30.     @ApiOperation(value = "备份交换机发送一条确认的消息")
  31.     @GetMapping("/sendMsg/{message}")
  32.     public void sendMsg(@PathVariable String message) {
  33.         log.info("当前时间：{}，发送消息开始:{}", LocalDateTime.now(), message);
  35.         CorrelationData correlationData = new CorrelationData("111");
  36.         rabbitTemplate.convertAndSend(BackupExchangeConfig.NORMAL_EXCHANGE, BackupExchangeConfig.NORMAL_ROUTING_KEY, BackupExchangeConfig.NORMAL_ROUTING_KEY + ":" + message, correlationData);
  38.         /**
  39.          * 发送到一个假的exchange上
  40.          * 消息ack为false
  41.          */
  42.         CorrelationData correlationData2 = new CorrelationData("222");
  44.         rabbitTemplate.convertAndSend("FAKE_EXCHANGE", BackupExchangeConfig.NORMAL_EXCHANGE, BackupExchangeConfig.NORMAL_ROUTING_KEY + ":" + message, correlationData2);
  46.         /**
  47.          * 发送到一个假的routingKey上
  48.          * 设置了备份交换机 消息不会被退回 （备份交换机的级别比设置退回消息权限高）
  49.          */
  50.         CorrelationData correlationData3 = new CorrelationData("333");
  51.         rabbitTemplate.convertAndSend(BackupExchangeConfig.NORMAL_EXCHANGE, "FAKE_ROUTING_KEY", "FAKE_ROUTING_KEY" + ":" + message, correlationData3);
  54.         log.info("当前时间：{}，发送消息完毕:{}", LocalDateTime.now(), message);
  55.     }
  58. }

九、rabbitMQ中其他知识点

9.1 幂等性

用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用。举个最简单的例子，那就是支付，用户购买商品后支付，支付扣款成功，但是返回结果的时候网络异常，此时钱已经扣了，用户再次点击按钮，此时会进行第二次扣款，返回结果成功，用户查询余额发现多扣钱了，流水记录也变成了两条。

解决思路

  • redis天然具有幂等性 可以使用redis执行setnx,从而实现不重复消费
  • 唯一id加指纹码机制，在分布式系统中生成唯一的mq id 来实现

    9.2 优先级队列

    ```java package com.rem.rabbitmq.boot.Dpriority;

import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;

/**

  • 优先级队列 *
  • @author Rem
  • @date 2021-12-29 */

@Configuration public class PriotiryConfig {

  1. public static final String PRIOTIRY_EXCHANGE = "PRIOTIRY_EXCHANGE";
  3. public static final String PRIOTIRY_ROUTING_KEY = "PRIOTIRY_ROUTING_KEY";
  5. public static final String PRIOTIRY_QUEUE = "PRIOTIRY_QUEUE";
  7. @Bean("priotiryExchange")
  8. public DirectExchange priotiryExchange() {
  9.     return ExchangeBuilder.directExchange(PRIOTIRY_EXCHANGE).build();
  10. }
  12. /**
  13.  * 设置优先级 0~255 官网推荐1-10 设置高比较吃内存和cpu
  14.  * 发送的消息也必须设置优先级
  15.  *
  16.  * @return
  17.  */
  18. @Bean("priotiryQueue")
  19. public Queue priotiryQueue() {
  20.     return QueueBuilder.durable(PRIOTIRY_QUEUE).maxPriority(5).build();
  21. }
  23. @Bean
  24. public Binding priotiryQueueaBinding(@Qualifier("priotiryQueue") Queue priotiryQueue, @Qualifier("priotiryExchange") DirectExchange priotiryExchange) {
  25.     return BindingBuilder.bind(priotiryQueue).to(priotiryExchange).with(PRIOTIRY_ROUTING_KEY);
  26. }

}

  2. ```java
  3. package com.rem.rabbitmq.boot.Dpriority;
  5. import io.swagger.annotations.Api;
  6. import io.swagger.annotations.ApiOperation;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.springframework.amqp.core.MessagePostProcessor;
  9. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.web.bind.annotation.GetMapping;
  12. import org.springframework.web.bind.annotation.PathVariable;
  13. import org.springframework.web.bind.annotation.RequestMapping;
  14. import org.springframework.web.bind.annotation.RestController;
  16. import java.time.LocalDateTime;
  18. /**
  19.  * @author Rem
  20.  * @date 2021-12-30
  21.  */
  23. @Slf4j
  24. @RestController
  25. @RequestMapping("/priotiry")
  26. @Api(tags = "优先级队列")
  27. public class PriotiryController {
  29.     @Autowired
  30.     private RabbitTemplate rabbitTemplate;
  32.     @ApiOperation(value = "发送一条的消息并且遍历")
  33.     @GetMapping("/sendMsg/{message}")
  34.     public void sendMsg(@PathVariable String message) {
  36.         for (int i = 0; i < 10; i++) {
  37.             String msg = i + ":" + message;
  38.             log.info("当前时间：{}，发送消息开始:{}", LocalDateTime.now(), msg);
  39.             if (i % 3 == 0) {
  40.                 //给消息赋予一个priority属性
  41.                 Integer finalI = i;
  42.                 MessagePostProcessor messagePostProcessor = messageProperties -> {
  43.                     messageProperties.getMessageProperties().setPriority(finalI);
  44.                     return messageProperties;
  45.                 };
  46.                 rabbitTemplate.convertAndSend(PriotiryConfig.PRIOTIRY_EXCHANGE, PriotiryConfig.PRIOTIRY_ROUTING_KEY, msg, messagePostProcessor);
  47.             } else {
  48.                 rabbitTemplate.convertAndSend(PriotiryConfig.PRIOTIRY_EXCHANGE, PriotiryConfig.PRIOTIRY_ROUTING_KEY, msg);
  49.             }
  50.             log.info("当前时间：{}，发送消息完毕:{}", LocalDateTime.now(), msg);
  51.         }
  52.     }
  55. }
  1.  /**
  2.      * 优先级队列监听
  3.      *
  4.      * @param message
  5.      */
  6.     @RabbitListener(queues = PriotiryConfig.PRIOTIRY_QUEUE)
  7.     public void receivePriotiry(Message message) {
  8.         String msg = new String(message.getBody());
  9.         log.info("优先级队列收到的消息是:{}", msg);
  10.     }

9.3 惰性队列

  1. package com.rem.rabbitmq.boot.Elazy;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * 惰性队列
  10.  * 在发送1百万条消息，每条消息大概占1KB的情况下，普通队列占用内存是1.2GB，而惰性队列仅仅占用1.5MB
  11.  *
  12.  * @author Rem
  13.  * @date 2021-12-29
  14.  */
  16. @Configuration
  17. public class LazyConfig {
  19.     public static final String LAZY_EXCHANGE = "LAZY_EXCHANGE";
  21.     public static final String LAZY_ROUTING_KEY = "LAZY_ROUTING_KEY";
  23.     public static final String LAZY_QUEUE = "LAZY_QUEUE";
  26.     @Bean("lazyExchange")
  27.     public DirectExchange lazyExchange() {
  28.         return ExchangeBuilder.directExchange(LAZY_EXCHANGE).build();
  29.     }
  31.     /**
  32.      * 声明队列为惰性队列 消息直接存放在磁盘中
  33.      *
  34.      * @return
  35.      */
  36.     @Bean("lazyQueue")
  37.     public Queue lazyQueue() {
  38.         return QueueBuilder.durable(LAZY_QUEUE).lazy().build();
  39.     }
  41.     @Bean
  42.     public Binding lazyQueueaBinding(@Qualifier("lazyQueue") Queue lazyQueue, @Qualifier("lazyExchange") DirectExchange lazyExchange) {
  43.         return BindingBuilder.bind(lazyQueue).to(lazyExchange).with(LAZY_ROUTING_KEY);
  44.     }
  47. }
  1. package com.rem.rabbitmq.boot.Elazy;
  3. import io.swagger.annotations.Api;
  4. import io.swagger.annotations.ApiOperation;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.web.bind.annotation.GetMapping;
  9. import org.springframework.web.bind.annotation.PathVariable;
  10. import org.springframework.web.bind.annotation.RequestMapping;
  11. import org.springframework.web.bind.annotation.RestController;
  13. import java.time.LocalDateTime;
  15. /**
  16.  * @author Rem
  17.  * @date 2021-12-30
  18.  */
  20. @Slf4j
  21. @RestController
  22. @RequestMapping("/lazy")
  23. @Api(tags = "惰性队列")
  24. public class LazyController {
  26.     @Autowired
  27.     private RabbitTemplate rabbitTemplate;
  29.     @ApiOperation(value = "发送一条的消息")
  30.     @GetMapping("/sendMsg/{message}")
  31.     public void sendMsg(@PathVariable String message) {
  32.         log.info("当前时间：{}，发送消息开始:{}", LocalDateTime.now(), message);
  33.         rabbitTemplate.convertAndSend(LazyConfig.LAZY_EXCHANGE, LazyConfig.LAZY_ROUTING_KEY, message);
  34.         log.info("当前时间：{}，发送消息完毕:{}", LocalDateTime.now(), message);
  35.     }
  36. }
  1.  /**
  2.      * 惰性队列监听
  3.      *
  4.      * @param message
  5.      */
  6.     @RabbitListener(queues = LazyConfig.LAZY_QUEUE)
  7.     public void receiveLazy(Message message) {
  8.         String msg = new String(message.getBody());
  9.         log.info("惰性队列 的消息是:{}", msg);
  10.     }

十、rabbitMQ集群

10.1 docker 搭建rabbitMQ集群

创建三个子文件夹

  1. mkdir -p /home/rabbitmq
  2. cd /mydata/rabbitmq-cluster 
  3. mkdir rabbitmq-1 rabbitmq-2 rabbitmq-3

分别创建三个docker容器

  1. docker run --hostname rabbitmq-1 --name rabbitmq-1 \
  2. -p 15675:15672 \
  3. -p 5675:5672 \
  4. -v /home/rabbitmq/rabbitmq-1:/var/lib/rabbitmq \
  5. -e RABBITMQ_ERLANG_COOKIE='rabbitmq-cookie' \
  6. -id rabbitmq:management
  1. docker run --hostname rabbitmq-2 --name rabbitmq-2 \
  2. -p 15673:15672 \
  3. -p 5673:5672 \
  4. -v /home/rabbitmq/rabbitmq-2:/var/lib/rabbitmq \
  5. -e RABBITMQ_ERLANG_COOKIE='rabbitmq-cookie' \
  6. --link rabbitmq-1:rabbitmq-1 \
  7. -id rabbitmq:management
  1. docker run --hostname rabbitmq-3 --name rabbitmq-3 \
  2. -p 15674:15672 \
  3. -p 5674:5672 \
  4. -v /home/rabbitmq/rabbitmq-3:/var/lib/rabbitmq \
  5. -e RABBITMQ_ERLANG_COOKIE='rabbitmq-cookie' \
  6. --link rabbitmq-1:rabbitmq-1 \
  7. --link rabbitmq-2:rabbitmq-2 \
  8. -id rabbitmq:management

让节点加入集群

  1. docker exec -it rabbitmq-1 /bin/bash
  3. rabbitmqctl stop_app
  4. rabbitmqctl reset
  5. rabbitmqctl start_app
  1. docker exec -it rabbitmq-2 /bin/bash
  3. rabbitmqctl stop_app
  4. rabbitmqctl reset
  5. rabbitmqctl join_cluster --ram rabbit@rabbitmq-1
  6. rabbitmqctl start_app
  1. docker exec -it rabbitmq-1 /bin/bash
  3. rabbitmqctl stop_app
  4. rabbitmqctl reset
  5. rabbitmqctl start_app

访问任一节点 就可以看到集群情况了
image.png

10.2 镜像队列

引入镜像队列(Mirror Queue)的机制，可以将队列镜像到集群中的其他Broker节点之上，如果集群中的一个节点失效了，队列能自动地切换到镜像中的另一个节点上以保证服务的可用性。

自定义一个已rem 开头的 自定义镜像队列

image.png

创建结果
image.png

创建一个已rem开始的队列 在控制台就可以看到备份了一份，如果1号宕机，就会自动在三号再备份一份
image.png
image.png

10.3 高可用负载均衡

HAProxy提供高可用性、负载均衡及基于TCPHTTP应用的代理，支持虚拟主机，它是免费、快速并且可靠的一种解决方案，HAProxy实现了一种事件驱动、单一进程模型，此模型支持非常大的井发连接数。

如果前面配置的HAProxy主机突然宕机或者网卡失效，那么虽然RbbitMQ集群没有任何故障但是对于外界的客户端来说所有的连接都会被断开结果将是灾难性的为了确保负载均衡服务的可靠性同样显得十分重要，这里就要引入Keepalived它能够通过自身健康检查、资源接管功能做高可用(双机热备)，实现故障转移.

搭建效果图
1234.jpg