死信队列

介绍

死信，顾名思义就是无法被消费的消息。一般情况下，生产者生产的消息将直接发送给队列，消费者从队列中获取消息，但是如果出现某些原因可能会导致信息无法被消费，这样的消息如果没有后续的处理，就会变成死信。所以我们通过死信队列，来解决出现死信的问题。

应用场景：

为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。

还有比如说: 用户在商城下单成功并点击去支付后在指定时 间未支付时自动失效

死信的来源

• 消息 TTL 过期。
• 队列达到最大长度(队列满了，无法再添加数据到 MQ 中)。
• 消息被拒绝( basicReject() 或 basicNack() ，且方法的 requeue 参数为 false )。

以上三种情况消息都会成为死信，进入死信队列。

路由流程

死信Exchange路由死信消息的流程如下：

1. 生产者将消息发送到 Exchange。
2. Exchange 将消息路由到 Queue。
3. 消费者从 Queue 拉取消息。
4. 消息出现上小节的三种情况，消息成为死信。
5. Queue 根据 x-dead-letter-exchange 将死信消息发送到死信 Exchange，并根据 x-dead-letter-routing-key 为死信消息设置死信Routing Key。
6. 死信 Exchange 将死信消息路由到死信 Queue 。

所以在声明队列的时候，我们需要声明 x-dead-letter-exchange 和 x-dead-letter-routing-key 参数来定义死信消息的去向。

死信实战

代码架构图

生产者

在生产者的代码中，我们不需要做过多的事，只需要发送消息即可。

public class Producer {
    public final static String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        if (channel == null) {
            System.out.println("无法获取信道");
            return;
        }
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder()
                .expiration(TimeUnit.SECONDS.toMillis(10) + "")
                .build();
        // 循环发布消息
        for (int i = 0; i < 10; i++) {
            String msg = "hello " + i;
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, msg.getBytes(StandardCharsets.UTF_8));
            System.out.println("Producer 发布消息：" + msg);
        }
        RabbitMqUtils.closeAll(channel);
    }
}

正常消费者

public class Consumer01 {
    public final static String NORMAL_EXCHANGE = "normal_exchange";
    public final static String DEAD_EXCHANGE = "dead_exchange";
    public final static String NORMAL_QUEUE = "normal_queue";
    public final static String DEAD_QUEUE = "dead_queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        if (channel == null) {
            System.out.println("无法获取信道");
            return;
        }
        // 声明正常交换机和死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        // arguments 为正常队列需要的参数
        Map<String, Object> arguments = new HashMap<>();
        // 设置过期时间，但是我们一般不在声明交换机的时候设置，而是在发布消息的时候设置过期时间
        // arguments.put("x-message-ttl", TimeUnit.SECONDS.toMillis(10));
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信routingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        // 设置正常队列的长度
        // arguments.put("x-max-length", 6);
        // 声明正常队列和死信队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
        // 绑定正常交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
        System.out.println("Consumer01 等待接收消息...");
        // 接收消息回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String routingKey = message.getEnvelope().getRoutingKey();
            String msg = new String(message.getBody());
            if ("hello 2".equals(msg)) {
                System.out.println("来自 routingKey[" + routingKey + "] 的消息[已拒绝] -> " + msg);
                // 第二个参数需要为false，表示不要重新入队，才能进入到死信队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            } else {
                System.out.println("来自 routingKey[" + routingKey + "] 的消息 -> " + msg);
                // 手动应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        // 接收消息
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, (consumerTag, message) -> {
        });
    }
}

死信消费者

public class Consumer02 {
    public final static String DEAD_QUEUE = "dead_queue";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        if (channel == null) {
            System.out.println("无法获取信道");
            return;
        }
        System.out.println("Consumer02 等待接收消息...");
        // 接收消息回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String routingKey = message.getEnvelope().getRoutingKey();
            String msg = new String(message.getBody());
            System.out.println("来自 routingKey[" + routingKey + "] 的消息 -> " + msg);
            // 手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        // 接收消息
        channel.basicConsume(DEAD_QUEUE, false, deliverCallback, (consumerTag, message) -> {
        });
    }
}

测试

上面的代码，我们需要最先启动正常的消费者，正常消费者会创建两个交换机和两个队列，然后我们立即关闭正常消费者，模拟正常消费者突然宕机的情况，然后我们就可以开启生产者发布消息了：

发布消息后，我们开启死信消费者，等待10s后，消息过期就会成为死信，死信消费者应该可以全部获取到，这就是消息 TTL 过期的死信来源。

我们还可以取消正常消费者代码第 26 行的注释，让正常队列的长度为6，那么在生产者一次性发布10条消息，同时正常消费者宕机的情况下，死信队列就会收到多余的 4 条消息，这就是达到队列最大长度后的死信来源。

最后，我们依次开启正常消费者 -> 死信消费者 -> 生产者，在上面的代码中，我已经配置了当消息为 hello 2 的时候，消息会被正常消费者拒绝，并且不会 requeue ，在这种情况下，消息也会成为死信，这就是消息被拒绝的死信来源。

正常消费者：

死信消费者：

延迟队列

介绍

延时队列内部是有序的，最重要的特性就是体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理。简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

使用场景：

• 订单在十分钟之内未支付则自动取消。

• 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。

• 用户注册成功后，如果三天内没有登陆则进行短信提醒。

• 用户发起退款，如果三天内没有得到处理则通知相关运营人员。

• 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务。

看起来似乎我们不使用延时队列好像也可以完成上面的功能，我们只需要使用定时任务，一直轮询查询数据，每秒查一次，取出需要处理的数据进行处理就好了。其实对于数据量较小的情况，确实可以这样做。

但是对于数据量大，并且时效性要求比较高的场景，这样做就不太合适了，短期内未支付的订单数据可能会有很多，活动期间剩余成百上万的数据两，不可能在一秒内完成所有订单的检查，同时还会给数据库带来压力，无法满足业务要求而且性能低下，所以使用延时队列还是非常有必要的。

消息延迟

TTL 全称为 Time To Live，意味生存时间，在 RabbitMq 中消息的 TTL，也就是消息的最大存活时间，单位是毫秒。

我们可以给队列或者单独给某条消息设置 TTL：

• 如果给队列设置 TTL，那么整个队列中的所有消息都具有相同的存活时间。
• 如果给某条消息设置 TTL，那么只设置了该消息的存活时间。

如果同时配置了队列的 TTL 和消息的 TTL，那么较小的那个值将会被使用。当消息的存活时间达到 TTL 值时，若还没有被消费，则消息会成为死信。

消息设置 TTL

未使用 SpringBoot 方式：

AMQP.BasicProperties properties = new AMQP.BasicProperties()
    .builder()
    .expiration(TimeUnit.SECONDS.toMillis(10) + "")    // 设置消息的最大存活时间，单位毫秒
    .build();
channel.basicPublish(NORMAL_EXCHANGE    // 交换机
                     , "routingKey"        // routingKey
                     , properties        // 消息参数，即上面的properties对象
                     , msg.getBytes(StandardCharsets.UTF_8));    // 消息内容

整合 SpringBoot 方式：

rabbitTemplate.convertAndSend(RabbitMqConstants.X_NORMAL_EXCHANGE        // 交换机
                              , RabbitMqConstants.X_BINDING_QC            // routingKey
                              , msg.getBytes(StandardCharsets.UTF_8)    // 消息内容
                              , message -> {                            // 消息后置处理器
                                  // 设置消息过期时间，单位毫秒
                                  message.getMessageProperties().setExpiration(ttl);
                                  return message;
                              });

队列设置 TTL

未使用 SpringBoot 方式：

// arguments 为队列需要的参数
Map<String, Object> arguments = new HashMap<>();
// 设置过期时间，单位毫秒
arguments.put("x-message-ttl", TimeUnit.SECONDS.toMillis(10));
// 声明队列时需要传入参数
channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

整合 SpringBoot 方式：

@Bean
public Queue queue() {
    return QueueBuilder
        .durable(RabbitMqConstants.QUEUE_B)    // 队列需要持久化，并设置队列名
        .ttl(TimeUnit.SECONDS.toMillis(40))    // 队列的最大存活时间 
        .build();
}

两种方式的区别

使用队列方式设置 TTL，消息过期之后，则会被队列丢弃(若配置了死信队列，则进入死信队列)；但是使用的消息方式设置的 TTL，在消息过期之后可能并不会被立刻丢弃，这是因为消息是在即将投递给消费者的时候才判断是否过期的，那么即使消息过期，但是在这个消息之前的其他消息未被消费或过期，则该消息仍然积压在队列中。

关于这两种方式的区别，在后面实战部分可以更明显的感觉到，如果这里不理解可以待会儿在实战部分理解。

另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

整合 SpringBoot

引入依赖

<!--RabbitMQ 依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- springboot整合web启动器 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 阿里巴巴的json工具 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.47</version>
</dependency>
<!-- swagger相关依赖[可选] -->
<dependency>
    <groupId>io.springfox</groupId>
    <artifactId>springfox-swagger2</artifactId>
    <version>2.9.2</version>
</dependency>
<!-- swagger相关依赖[可选] -->
<dependency>
    <groupId>io.springfox</groupId>
    <artifactId>springfox-swagger-ui</artifactId>
    <version>2.9.2</version>
</dependency>
<!--RabbitMQ 测试依赖-->
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>

修改 application.yml 配置文件

spring:
  rabbitmq:
    host: 192.168.20.132
    port: 5672
    username: admin
    password: 123456

添加 Swagger 配置类[可选]

@Configuration
public class SwaggerConfig {
    @Bean
    public Docket docket() {
        return new Docket(DocumentationType.SWAGGER_2)
                .apiInfo(apiInfo())
                .groupName("WebApi")
                .select()
                .build();
    }
    //配置文档信息
    private ApiInfo apiInfo() {
        Contact contact = new Contact("留白", "http://www.liubaiblog.top", "123456789@qq.com");
        return new ApiInfo(
                "RabbitMQ 接口文档", // 标题
                "本文档描述了 RabbitMQ 微服务接口定义", // 描述
                "v1.0", // 版本
                null, // 组织链接
                contact, // 联系人信息
                null,
                null,
                new ArrayList<>()// 扩展
        );
    }
}

SpringBoot 整合实战

前一小节我们介绍了死信队列，刚刚又介绍了 TTL，至此利用 RabbitMQ 实现延时队列的两大要素已经集齐，接下来只需要将它们进行融合，再加入一点点调味料，延时队列就可以新鲜出炉了。

想想看，延时队列，不就是想要消息延迟多久被处理吗，TTL 则刚好能让消息在延迟多久之后成为死信，另一方面， 成为死信的消息都会被投递到死信队列里，这样只需要消费者一直消费死信队列里的消息就完事了，因为里面的消息都是希望被立即处理的消息。

代码架构图

创建两个队列 QA 和 QB，两者队列 TTL 分别设置为 10S 和 40S，然后在创建一个交换机 X 和死信交 换机 Y，它们的类型都是 direct，创建一个死信队列 QD，它们的绑定关系如下：

常量类

根据上面架构图的关系，我们先定义如下常量，方便后续的使用。

public class RabbitMqConstants {
    public final static String X_NORMAL_EXCHANGE = "X";
    public final static String Y_DEAD_EXCHANGE = "Y";
    public final static String QUEUE_A = "QA";
    public final static String QUEUE_B = "QB";
    public final static String QUEUE_D = "QD";
    public final static String X_BINDING_QA = "XA";
    public final static String X_BINDING_QB = "XB";
    public final static String Y_BINDING_QD = "YD";
    public final static int QA_TTL = (int) TimeUnit.SECONDS.toMillis(10);
    public final static int QB_TTL = (int) TimeUnit.SECONDS.toMillis(40);
}

配置类

之前声明交换机和队列的工作，我们都是交给消费者执行的，但是 SpringBoot 整合之后，我们可以直接在配置类中声明，不用那么麻烦。

在配置类中：

• XxxExchange ：表示声明一个交换机。
• Queue ：表示声明一个队列，注意包的路径为 org.springframework.amqp.core 。
• Binding ：声明一个绑定关系。

代码如下：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import top.liubaiblog.springbootrabbitmq.constant.RabbitMqConstants;
@Configuration
public class RabbitMqConfig {
    @Bean   // 声明x正常交换机
    public DirectExchange xExchange() {
        return new DirectExchange(RabbitMqConstants.X_NORMAL_EXCHANGE);
    }
    @Bean   // 声明y死信交换机
    public DirectExchange yExchange() {
        return new DirectExchange(RabbitMqConstants.Y_DEAD_EXCHANGE);
    }
    @Bean   // 声明队列A，设置延迟10s
    public Queue queueA() {
        return QueueBuilder
                .durable(RabbitMqConstants.QUEUE_A)
                .ttl(RabbitMqConstants.QA_TTL)
                .deadLetterExchange(RabbitMqConstants.Y_DEAD_EXCHANGE)
                .deadLetterRoutingKey(RabbitMqConstants.Y_BINDING_QD)
                .build();
    }
    @Bean   // 声明队列B，设置延迟40s
    public Queue queueB() {
        return QueueBuilder
                .durable(RabbitMqConstants.QUEUE_B)
                .ttl(RabbitMqConstants.QB_TTL)
                .deadLetterExchange(RabbitMqConstants.Y_DEAD_EXCHANGE)
                .deadLetterRoutingKey(RabbitMqConstants.Y_BINDING_QD)
                .build();
    }
    @Bean   // 声明死信队列D
    public Queue queueD() {
        return QueueBuilder
                .durable(RabbitMqConstants.QUEUE_D)
                .build();
    }
    @Bean   // 绑定X交换机和队列A
    public Binding xExchangeBindingQa() {
        return BindingBuilder
                .bind(queueA())
                .to(xExchange())
                .with(RabbitMqConstants.X_BINDING_QA);
    }
    @Bean   // 绑定X交换机和队列B
    public Binding xExchangeBindingQb() {
        return BindingBuilder
                .bind(queueB())
                .to(xExchange())
                .with(RabbitMqConstants.X_BINDING_QB);
    }
    @Bean   // 绑定Y死信交换机和死信队列D
    public Binding yExchangeBindingQd() {
        return BindingBuilder
                .bind(queueD())
                .to(yExchange())
                .with(RabbitMqConstants.Y_BINDING_QD);
    }
}

生产者

浏览器访问 http://localhost:8080/ttl/sendMsg/{msg} 接口，即可生产消息。

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PostMapping("/sendMsg/{msg}")
    public String sendMsg(@PathVariable("msg") String msg) {
        log.info("[{}] 发送消息 -> {}", new Date(), msg);
        rabbitTemplate.convertAndSend(RabbitMqConstants.X_NORMAL_EXCHANGE    // 交换机
                , RabbitMqConstants.X_BINDING_QA                            // routingKey
                , msg.getBytes(StandardCharsets.UTF_8));                    // 消息
        rabbitTemplate.convertAndSend(RabbitMqConstants.X_NORMAL_EXCHANGE
                , RabbitMqConstants.X_BINDING_QB
                , msg.getBytes(StandardCharsets.UTF_8));
        return "success";
    }
}

消费者

消费者的方法中需要标识 @RabbitListener 注解，表示这个方法监听某个队列中的消息。

@Slf4j
@Component
public class DeadLetterQueueConsumer {
    @RabbitListener(queues = RabbitMqConstants.QUEUE_D)
    public void receiveD(Message message, Channel channel) {
        String body = new String(message.getBody());
        log.info("[{}] 接收到死信队列消息 -> {}", new Date(), body);
    }
}

测试

发送请求：http://localhost:8080/ttl/sendMsg/helloworld 。

第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 40S 之后变成了死信消息， 然后被消费掉，这样一个延时队列就打造完成了。

不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列？这里只有 10S 和 40S 两个时间选项，如果需要一个小时后处理，那么就需要增加 TTL 为一个小时的队列，如果是预定会议室然 后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？

显然，这样是不太合适的，我们还需要对延迟队列进行优化。

优化

在这里新增了一个队列 QC，绑定关系如下，该队列不设置 TTL 时间：

但是队列不设置 TTL 时间，我们可以将消息设置 TTL 时间，这样我们就可以任意的指定消息什么时候过期。

常量类：

public class RabbitMqConstants {
    // ...
    public final static String QUEUE_C = "QC";
    public final static String X_BINDING_QC = "XC";
}

配置类：

@Configuration
public class RabbitMqConfig {
       // ...
    @Bean
    public Queue queueC() {
        return QueueBuilder
                .durable(RabbitMqConstants.QUEUE_C)
                .deadLetterExchange(RabbitMqConstants.Y_DEAD_EXCHANGE)
                .deadLetterRoutingKey(RabbitMqConstants.Y_BINDING_QD)
                .build();
    }
    @Bean
    public Binding xExchangeBindingQc() {
        return BindingBuilder
                .bind(queueC())
                .to(xExchange())
                .with(RabbitMqConstants.X_BINDING_QC);
    }
}

生产者：

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // ...
    @PostMapping("/sendTtlMsg/{msg}/{ttlSeconds}")
    public String sendTtlMsg(@PathVariable String msg, @PathVariable Integer ttlSeconds) {
        log.info("[{}] 发送消息并延迟{}s -> {}", new Date(), ttlSeconds, msg);
        String ttl = TimeUnit.SECONDS.toMillis(ttlSeconds) + "";
        rabbitTemplate.convertAndSend(RabbitMqConstants.X_NORMAL_EXCHANGE    // 交换机
                , RabbitMqConstants.X_BINDING_QC                            // routingKey
                , msg.getBytes(StandardCharsets.UTF_8)                        // 消息
                , message -> {                                                // 消息后置处理器
                    // 设置消息过期时间
                    message.getMessageProperties().setExpiration(ttl);
                    return message;
                });
        return "success";
    }
}

问题

依次发送请求：

• http://localhost:8080/ttl/sendTtlMsg/hello1/20
• http://localhost:8080/ttl/sendTtlMsg/hello2/2

我的本意是想要 hello1 消息在延时 20s 后被消费，hello2 消息在延时 2s 后被消费，但是看下图结果，就会发现 hello2 依然延迟了 20s 才被消费。

这就是出现的问题，还记得我们在消息延迟小节说队列 TTL 和消息 TTL 的区别吗，我们使用消息 TTL 之后，消息只有在即将投递的时候才会判断是否过期，但是在 hello2 消息之前，我们还有 hello1 消息没有被消费，所以导致 hello2 消息即不会被消费，更不会被判断是否过期，从而积压在队列中。

等到 20s 后，hello1 消息过期，被投递到死信队列，然后才到 hello2 ，hello2 再被判定为过期，也转入死信队列，从而出现上图的结果。

但是很明显上图的结果不是我们想要的，我们想要的结果是 hello2 2s 后先被消费，然后 hello1 20s 后再被消费。

RabbitMq 插件实现延迟队列

为了解决上面的问题，我们可以使用 RabbitMq 为我们提供的插件。RabbitMq 支持很多插件，我们可以访问其官网获取，我们这次需要的是 rabbitmq_delayed_message_exchange 插件，访问下面地址可以直接跳转到插件的 GitHub 下载。

获取插件

插件地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases。

在 Linux 中，进入如下目录，并将 ez 文件放入到此目录下：

$ cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.10.1/plugins/

执行如下命令安装插件

$ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

如果 Web 管理界面后，有如下效果表示安装成功，如果没有出现可以尝试重启服务。

代码架构图

使用了延迟插件之后，我们可以选择 x-delayed-message 类型的交换机，这个方式和之前的很不同，在之前，我们其实是基于的死信队列实现的延迟队列，是等待消息过期之后进入死信队列后，我们再消费死信队列的消息。

但是使用延迟插件后，我们可以不借助死信队列，消息的延迟在交换机位置，消息会在交换机延迟到指定时间后，再转发给队列，然后消费者可以直接消费队列中的消息。

在这里新增了一个队列 delayed.queue ，一个自定义交换机 delayed.exchange ，我们下面根据上图编写代码。

常量类

public class RabbitMqConstants {
    public final static String DELAY_EXCHANGE = "delayed.exchange";
    public final static String DELAY_QUEUE = "delayed.queue";
    public final static String DELAY_EXCHANGE_BINDING_DELAY_QUEUE = "delayed.routingKey";
}

配置类

注意，由于这里的交换机不是之前的几种常见类型的交换机，所以在声明交换机的时候，我们需要使用 CustomExchange 类，表示这是一个自定义交换机。

@Configuration
public class DelayRabbitMqConfig {
    @Bean   // 声明自定义交换机
    public CustomExchange delayExchange() {
        Map<String, Object> arguments = new HashMap<>(3);
        arguments.put("x-delayed-type", "direct");
        return new CustomExchange(RabbitMqConstants.DELAY_EXCHANGE  // 交换机名称
                , "x-delayed-message"                          // 自定义消息类型
                , true                                      // 是否持久化
                , false                                  // 是否自动删除
                , arguments);                                       // 参数列表
    }
    @Bean   // 声明延迟队列
    public Queue delayQueue() {
        return QueueBuilder
                .durable(RabbitMqConstants.DELAY_QUEUE)
                .build();
    }
    @Bean   // 声明routingKey
    public Binding delayExchangeBindingDelayQueue() {
        return BindingBuilder
                .bind(delayQueue())
                .to(delayExchange())
                .with(RabbitMqConstants.DELAY_EXCHANGE_BINDING_DELAY_QUEUE)
                .noargs();
    }
}

生产者

生产者的代码和之前基本一致，但是我们这里使用另一种方式发送，这种方式的效果和之前是一致的。但是在 MessageProperties 中，不是使用 setExpiration() 方法，而是使用 setDelay() 方法表示设置延迟时间。

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PostMapping("/sendDelayMsg/{msg}/{ttlSeconds}")
    public String sendDelayMsg(@PathVariable String msg, @PathVariable Integer ttlSeconds) {
        log.info("[{}] 发送自定义延迟队列消息并延迟{}s -> {}", new Date(), ttlSeconds, msg);
        int ttl = (int) TimeUnit.SECONDS.toMillis(ttlSeconds);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDelay(ttl);
        Message message = new Message(msg.getBytes(StandardCharsets.UTF_8), messageProperties);
        rabbitTemplate.send(RabbitMqConstants.DELAY_EXCHANGE
                , RabbitMqConstants.DELAY_EXCHANGE_BINDING_DELAY_QUEUE
                , message);
        return "success";
    }
}

消费者

@Slf4j
@Component
public class DeadLetterQueueConsumer {
    @RabbitListener(queues = RabbitMqConstants.DELAY_QUEUE)
    public void receiveDelayQueue(Message message) {
        String body = new String(message.getBody());
        log.info("[{}] 接收到自定义延迟队列消息 -> {}", new Date(), body);
    }
}

测试

依次发送请求：

• http://localhost:8080/ttl/sendDelayMsg/hello1/20
• http://localhost:8080/ttl/sendDelayMsg/hello2/2

从这次的结果来看，是符合我们的预期的，我们是希望 hello2 消息延迟 2s 后就交给消费者消费，结果也确实是这样。

阶段总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用 RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。

另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz 或者利用 kafka 的时间轮，这些方式各有特点，看需要适用的场景。

发布确认高级

在生产环境中由于一些不明原因，导致 rabbitmq 重启，在 RabbitMQ 重启期间生产者消息投递失败， 导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？ 特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢？

为了方便，我们还是使用 SpringBoot 的方式对 RabbitMq 整合。

发布确认

代码架构图

application.yml 配置文件

除了 rabbitMq 地址相关的信息外，我们还需要在配置文件中添加如下内容：

spring:
  rabbitmq:
    publisher-confirm-type: correlated

publisher-confirm-type 一共有三个值：

• NONE ：禁用发布确认模式，是默认值。
• CORRELATED ：发布消息成功到交换器后会触发回调方法。
• SIMPLE ：经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法；其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms() 或 waitForConfirmsOrDie 方法 等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker。

常量类

public class RabbitMqConstants {
    public final static String CONFIRM_EXCHANGE = "confirm.exchange";
    public final static String CONFIRM_QUEUE = "confirm.queue";
    public final static String EXCHANGE_BINDING_QUEUE = "confirm.eq";
}

配置类

@Configuration
public class RabbitMqConfig {
    @Bean   // 声明交换机
    public DirectExchange confirmExchange() {
        return new DirectExchange(RabbitMqConstants.CONFIRM_EXCHANGE);
    }
    @Bean   // 声明队列
    public Queue confirmQueue() {
        return QueueBuilder
                .durable(RabbitMqConstants.CONFIRM_QUEUE)
                .build();
    }
    @Bean   // 声明绑定关系
    public Binding binding() {
        return BindingBuilder
                .bind(confirmQueue())
                .to(confirmExchange())
                .with(RabbitMqConstants.EXCHANGE_BINDING_QUEUE);
    }
}

生产者

在生产者这里的代码需要有些不同，之前的方法中我们是没有指定消息的ID的，但是其实我们生产的消息应该携带一个消息ID的，携带这个消息ID我们可以通过创建一个 CorrelationData 类实现，如下。

@Slf4j
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 请求这个接口，我们发送两个消息，一个给存在的交换机，一个给不存在的交换机
    @PostMapping("/send/{msg}")
    public String sendMsg(@PathVariable String msg) {
        // 给存在的交换机发送消息
        rabbitTemplate.convertAndSend(RabbitMqConstants.CONFIRM_EXCHANGE
                , RabbitMqConstants.EXCHANGE_BINDING_QUEUE
                , msg.getBytes(StandardCharsets.UTF_8)
                , new CorrelationData("1"));
        // 给不存在的交换机发送消息
        rabbitTemplate.convertAndSend(RabbitMqConstants.CONFIRM_EXCHANGE + "abcd"
                , RabbitMqConstants.EXCHANGE_BINDING_QUEUE
                , msg.getBytes(StandardCharsets.UTF_8)
                , new CorrelationData("2"));
        log.info("发送了消息 -> {}", msg);
        return "success";
    }
}

消费者

@Slf4j
@Component
public class ConsumerListener {
    @RabbitListener(queues = RabbitMqConstants.CONFIRM_QUEUE)
    public void confirm(Message message) throws IOException {
        String body = new String(message.getBody());
        log.info("消费者接收到消息 -> {}", body);
    }
}

消息确认的回调接口

我们可以定义一个用于消息确认的回调接口，这个回调接口不管在消息接收到还是没有接收到都会被触发，通过 ack 属性可以判断消息是否被接收到。

这个回调接口要求继承 RabbitTemplate.ConfirmCallback 接口，这个是个函数式接口，我们也可以不单独定义类，而是使用 Lamdba 表达式，但是本例还是使用声明一个新类的方式。

最后我们需要把我们声明的回调接口，注入 rabbitTemplate 当中，这样才能生效。

@Component
@Slf4j
public class MessageCallback implements RabbitTemplate.ConfirmCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 初始化方法，在构造器和rabbitTemplate注入之后执行
    @PostConstruct
    private void init() {
        // 将本类注入到rabbitTemplate
        rabbitTemplate.setConfirmCallback(this);
    }
    /**
     * 消息确认
     *
     * @param correlationData 相关数据，这个需要生产者给出
     * @param ack             是否应答
     * @param errorCause      错误原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String errorCause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机收到id为{}的消息", id);
        } else {
            log.info("交换机未收到id为{}的消息，由于原因 -> {}", id, errorCause);
        }
    }
}

结果

访问地址 http://localhost:8081/producer/send/hello ，会发送两个消息，一个 ID 为 1，一个 ID 为 2，很明显，由于我们在生产者端声明了一个不存在的交换机，所以 ID 为 2 的消息必然会收不到。

但是上面的效果就已经够了吗？其实我们可以尝试一下生产者端向一个不存在的 routingKey 发送消息，这个时候，交换机无法通过 routingKey 找到对应的队列投递消息，就会把消息丢弃，但是在回调接口中，并不会体现出来，但实际上消费者只接受到了一个消息。

回退消息

Mandatory 参数

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。

那么如何让无法被路由的消息帮我想办法处理一下？最起码通知我一声，我好自己处理啊。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

开启这个参数的方式有两种，一种是在配置文件中开启，一种是在 rabbitTemplate 中设置此参数为 true 。

配置文件

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true

rabbitTemplate

rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnsCallback(<消息回退的接口>);

生产者

@Slf4j
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 请求这个接口，我们发送两个消息，一个给存在的交换机，一个给不存在的交换机
    @PostMapping("/send/{msg}")
    public String sendMsg(@PathVariable String msg) {
        // 给存在的交换机发送消息
        rabbitTemplate.convertAndSend(RabbitMqConstants.CONFIRM_EXCHANGE
                , RabbitMqConstants.EXCHANGE_BINDING_QUEUE
                , msg.getBytes(StandardCharsets.UTF_8)
                , new CorrelationData("1"));
        // 给不存在的交换机发送消息
        rabbitTemplate.convertAndSend(RabbitMqConstants.CONFIRM_EXCHANGE
                , RabbitMqConstants.EXCHANGE_BINDING_QUEUE + "dada"
                , msg.getBytes(StandardCharsets.UTF_8)
                , new CorrelationData("2"));
        log.info("发送了消息 -> {}", msg);
        return "success";
    }
}

消息回退接口

@Component
@Slf4j
public class MessageCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 初始化方法，在构造器和rabbitTemplate注入之后执行
    @PostConstruct
    private void init() {
        // 将本类注入到rabbitTemplate
        rabbitTemplate.setConfirmCallback(this);
        // 如果在配置文件中配置类Mandatory参数，这里可以不设置
        // rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(this);
    }
    /**
     * 消息确认
     *
     * @param correlationData 相关数据，这个需要生产者给出
     * @param ack             是否应答
     * @param errorCause      错误原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String errorCause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机收到id为{}的消息", id);
        } else {
            log.info("交换机未收到id为{}的消息，由于原因 -> {}", id, errorCause);
        }
    }
    /**
     * 处理被回退的消息
     *
     * @param returnedMessage 被回退的消息
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        String msg = new String(returnedMessage.getMessage().getBody());
        String cause = returnedMessage.getReplyText();
        String routingKey = returnedMessage.getRoutingKey();
        log.error("消息 [{}] 被交换机退回，退回原因：{}，路由key为{}", msg, cause, routingKey);
    }
}

结果

访问地址 http://localhost:8081/producer/send/hello ，生产者发布了向不存在的 routingKey 发送消息，消息会被退回。

备份交换机

设置 Mandatory 参数后，可以实现回退消息，但是还有一个弊端就是对回退的消息，我们应该怎么处理呢？我们可以使用日志打印，但其实使用日志的方式处理是一种很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。

如果既不想丢失消息，也不想增加生产者的复杂性，我们就可以考虑使用备份交换机，备份交换机就相当于主交换机的备胎，当主交换机遇到无法路由的消息的时候，就会交给备份交换机处理，备份交换机通常是 fanout 类型，可以把收到的消息发布给所有绑定的队列，所以我们也可以创建多个消费者，对这个异常的消息进行处理。

比如我们可以建立一个报警队列，用独立的消费者进行监测和报警。

代码架构图

常量类

public class RabbitMqConstants {
    // ...
    public final static String BACKUP_EXCHANGE = "backup.exchange";
    public final static String BACKUP_QUEUE = "backup.queue";
    public final static String WARNING_QUEUE = "warning.queue";
    // ...
}

配置类

@Configuration
public class RabbitMqConfig {
    @Bean   //! 声明交换机
    public DirectExchange confirmExchange() {
        return ExchangeBuilder
                .directExchange(RabbitMqConstants.CONFIRM_EXCHANGE)     // 直接交换机
                .durable(true)                                          // 是否持久化
                .alternate(RabbitMqConstants.BACKUP_EXCHANGE)           // 备份交换机的名字
                .build();
    }
    @Bean   // 声明队列
    public Queue confirmQueue() {
        return QueueBuilder
                .durable(RabbitMqConstants.CONFIRM_QUEUE)
                .build();
    }
    @Bean   // 声明绑定关系
    public Binding binding() {
        return BindingBuilder
                .bind(confirmQueue())
                .to(confirmExchange())
                .with(RabbitMqConstants.EXCHANGE_BINDING_QUEUE);
    }
    @Bean   // 声明备份交换机，备份交换机采用广播方式发送消息，类型为fanout
    public FanoutExchange backupExchange() {
        return new FanoutExchange(RabbitMqConstants.BACKUP_EXCHANGE);
    }
    @Bean   // 备份队列
    public Queue backupQueue() {
        return QueueBuilder
                .durable(RabbitMqConstants.BACKUP_QUEUE)
                .build();
    }
    @Bean   // 备份报警队列
    public Queue warningQueue() {
        return QueueBuilder
                .durable(RabbitMqConstants.WARNING_QUEUE)
                .build();
    }
    @Bean   // 绑定备份交换机和备份队列
    public Binding backupExchangeBindingBackupQueue() {
        return BindingBuilder
                .bind(backupQueue())
                .to(backupExchange());
    }
    @Bean   // 绑定备份交换机和报警队列
    public Binding backupExchangeBindingWarningQueue() {
        return BindingBuilder
                .bind(warningQueue())
                .to(backupExchange());
    }
}

消费者

@Slf4j
@Component
public class ConsumerListener {
    // 正常消费者
    @RabbitListener(queues = RabbitMqConstants.CONFIRM_QUEUE)
    public void confirm(Message message) throws IOException {
        String body = new String(message.getBody());
        log.info("消费者接收到消息 -> {}", body);
    }
    // 备份的消费者
    @RabbitListener(queues = RabbitMqConstants.BACKUP_QUEUE)
    public void backup(Message message) {
        String body = new String(message.getBody());
        log.info("备份消费者接收到消息 -> {}", body);
    }
    // 报警的消费者
    @RabbitListener(queues = RabbitMqConstants.WARNING_QUEUE)
    public void warning(Message message) {
        String body = new String(message.getBody());
        log.error("警告 -> 消息 [{}] 未被正常消费", body);
    }
}

结果

Tip：在开始测试之前，需要把原来的 confirm.exchange 交换机删除，因为在配置类中我们添加了新的配置，不然程序会报错！

访问地址 http://localhost:8081/producer/send/hello ，如果出现不可路由的消息，交给交给备份交换机处理。

当 mandatory 参数和备份交换机一起使用的时候，默认备份交换机的优先级更高！

其他相关知识点

幂等性

介绍

每个队列的消息，都应该保证只被消费者消费一次，但是在某些情况下，消息可能会被多次消费。

比如：

• 消费者在消费完之后就挂掉了，但是还没有给 MQ 回复 ACK 消息，MQ 并不知道消费者已经挂掉了，等消费者恢复后，消息还会投递原消息给这个消费者，造成重复消费。
• 生产者在发布确认模式下，当生产者向 MQ 发布消息后，会等待 MQ 的确认，但是在这个时候网络延迟较高，那么生产者可能会重新发布这个消息，那么 MQ 就收到了两条相同的消息供消费者消费。

所以我们必须要保证同一个消息只会被消费一次，这就是幂等性问题。

解决思路

生产者每次发送消息之前，先为每条消息设置一个唯一ID，消费者拿到这个消息后，在消费之前，需要先进行判断，判断这个消息是否被消费，如果这个消息被消费了就不再进行消费。

所以问题就是怎么进行判断，比较常见的方式是消费后，将消息的唯一ID存到 SQL 数据库中 NoSQL 数据库中。

使用 SQL 数据库如 MySQL 的劣势就是在高并发时，如果是单个数 据库就会有写入性能瓶颈当然也可以采用分库分表提升性能，但也不是我们最推荐的方式。

更常见的方式是使用 Redis 数据库，这种数据库效率比较高，而且有 setnx 这类命令，可以直接帮助我们判断消息是否被消费过；或者也可以使用 Set 集合保存消费过的唯一ID。

实战

本例中，我们使用 redis 的方式演示，关于在 SpringBoot 中如何整合 redis，在我之前的文档中已经演示过，下面就直接开始演示。

对于在 SpringBoot 中如何声明交换机和队列并绑定，配置类已经在前面写过很多遍，这里也不再演示，我们主要聚焦生产者和消费者的代码，主要是消费者，在消费者中，我们需要保证消息仅被消费一次。

生产者

@Slf4j
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 请求这个接口，我们发送两个消息，一个给存在的交换机，一个给不存在的交换机
    @PostMapping("/send/{msg}")
    public String sendMsg(@PathVariable String msg) {
        // 生成消息
        Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8))
                .setMessageId(UUID.randomUUID().toString())
                .build();
        // 发送消息
        rabbitTemplate.send(RabbitMqConstants.CONFIRM_EXCHANGE
                , RabbitMqConstants.EXCHANGE_BINDING_QUEUE
                , message);
        return "success";
    }
}

消费者

Tip：在 SpringBoot 中，默认是 @RabbitListener 注解的方法结束后自动回复 ACK 给 MQ。

@Slf4j
@Component
public class ConsumerListener {
    @Autowired
    private RedisUtils redisUtils;
    @RabbitListener(queues = RabbitMqConstants.CONFIRM_QUEUE)
    public void confirm(Message message) throws Exception {
        String messageId = message.getMessageProperties().getMessageId();
        // 从redis中判断消息是否已经被消费
        boolean exists = redisUtils.setIsMember("rabbitMsg", messageId);
        if (exists) {
            log.warn("检测到重复消费的消息，消息ID -> {}", messageId);
            return;
        }
        // 消费消息
        String body = new String(message.getBody());
        log.info("消费者接收到消息ID为 [{}] 的消息 -> {}", messageId, body);
        // 将已经消费的消息ID保存到set集合
        redisUtils.setAdd("rabbitMsg", messageId);
        // 模拟长网络延迟
        TimeUnit.SECONDS.sleep(20);
    }
}

测试

为了演示出效果，我们先访问 http://localhost:8081/producer/send/hello1 地址向服务器发送 hello1 消息，在消息消费完之后，上面的代码中应该模拟了 20s 的网络延迟，在这个期间，我们直接强制关闭 SpringBoot 服务器。

然后我们再重启服务器，这时候，由于还没有恢复 ACK 消息，RabbitMQ 会认为消费者没有消费此消息，再次投递此消息给消费者，但是我们通过 redis 的判断，就避免了消息被重复消费的问题。

优先队列

介绍

优先队列就是具有优先级的队列，消息被消费的顺序根据消息的优先级决定，优先级高的消息先被消费。

假设我们系统有一个订单催付的功能，我们的客户在天猫下的订单，淘宝会及时将订单推送给我们，如果在用户设定的时间内未付款那么就会给用户推送一条短信提醒，功能似乎并不复杂。

但是这个功能中，肯定有大客户小客户，对于大客户来说，可以对我们提供较大的利润，自然需要优先对待，而小客户迟一些催付也无妨，在这种情况下，我们就可以使用优先队列的功能。

实战

常量类

public class RabbitMqConstants {
    public final static String PRIORITY_EXCHANGE = "priority_exchange";
    public final static String PRIORITY_QUEUE = "priority_queue";
    public final static String PRIORITY_EXCHANGE_BINDING_PRIORITY_QUEUE = "priority_eq";
}

配置类

在声明队列时，我们应该设置优先级的最高值，不建议设置太高，不然会导致系统负载太大。

@Configuration
public class PriorityQueueConfig {
    @Bean    // 声明交换机
    public DirectExchange priorityExchange() {
        return new DirectExchange(RabbitMqConstants.PRIORITY_EXCHANGE);
    }
    @Bean    // 声明优先队列
    public Queue priorityQueue() {
        return QueueBuilder
                .nonDurable(RabbitMqConstants.PRIORITY_QUEUE)
                .maxPriority(10)        //! 优先队列的最高值
                .build();
    }
    @Bean    // 声明绑定关系
    public Binding priorityBinding() {
        return BindingBuilder
                .bind(priorityQueue())
                .to(priorityExchange())
                .with(RabbitMqConstants.PRIORITY_EXCHANGE_BINDING_PRIORITY_QUEUE);
    }
}

生产者

在生产者中，发送的消息应该带有优先级，如果没有设置消息的优先级，则消息的优先级默认为0。

@Slf4j
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PostMapping("/send/priority/{msg}/{priority}")
    public String sendMsgPriority(@PathVariable String msg, @PathVariable Integer priority) {
        // 生成消息
        Message message = MessageBuilder
                .withBody(msg.getBytes(StandardCharsets.UTF_8))
                .setPriority(priority)
                .build();
        // 发送消息
        rabbitTemplate.send(RabbitMqConstants.PRIORITY_EXCHANGE
                , RabbitMqConstants.PRIORITY_EXCHANGE_BINDING_PRIORITY_QUEUE
                , message);
        return "success";
    }
}

消费者

@Slf4j
@Component
public class ConsumerListener {
    @RabbitListener(queues = RabbitMqConstants.PRIORITY_QUEUE)
    public void consume(Message message) {
        String body = new String(message.getBody());
        log.info("消费者接收到消息 -> {}", body);
    }
}

测试

Tip：在测试之前建议先注释消费者部分代码，因为如果开启消费者监听器，则发布一条消息就会被立即消费，所以建议先关闭。

依次访问如下地址：

• /producer/send/priority/hello/5
• /producer/send/priority/java/2
• /producer/send/priority/world/7
• /producer/send/priority/python/4

访问了这些地址之后，应该会向队列中推送四条消息，并有不同的优先级，预期优先级顺序是 world -> hello -> python -> java 。我们这时在取消消费者部分代码的注释，重新启动服务，可以看到控制台打印如下信息：

也确实符合我们的预期。

惰性队列

介绍

RabbitMQ 从 3.6.0 版本开始引入了惰性队列的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持 更多的消息存储。

当消费者由于各种各样的原因(比如消费者下线、宕机亦或者是由于维护而关闭等)而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。

默认情况下，当生产者将消息发送到 RabbitMQ 的时候，队列中的消息会尽可能的存储在内存之中， 这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当 RabbitMQ 需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然 RabbitMQ 的开发者们一直在升级相关的算法， 但是效果始终不太理想，尤其是在消息量特别大的时候。

两种模式

RabbitMQ 中队列具备两种模式：

• default
• lazy

默认使用的是 default 模式，在 3.6 之前的版本无需做任何变更；lazy 即为惰性队列的模式。

原生方式声明惰性队列

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");
channel.queueDeclare("testQueue", false, false, false, args);

SpringBoot 整合后声明惰性队列

@Bean
public Queue testQueue() {
    return QueueBuilder
        .nonDurable("testQueue")
        .lazy()
        .build();
}

内存开销对比

在发送 1 百万条消息，每条消息大概占 1KB 的情况下，普通队列占用内存是 1.2GB，而惰性队列仅仅占用 1.5MB。

集群

介绍

在前面使用的 RabbitMQ 都是在一台服务器上的，但是在真实的环境中，可能不能满足我们的需求。因为单台的服务器的服务是非常不安全的，一旦出现内存崩溃、服务器宕机等一系列问题，可能直接会导致无法传递消息，而出现这种问题对于应用程序来说可能是致命的。

为了避免出现上述的问题，我们可以将多台 RabbitMQ 服务器搭建成一个集群，在一个集群中，其中一个节点宕机，并不会影响其他节点的使用，可以很大程度上提高安全性。

并且搭建 RabbitMQ 集群可能很大程度上提高消息的吞吐量，在高吞吐量的场景中，搭建集群无疑也是一个很不错的选择。

搭建步骤

我们使用克隆的方式，先准备三台虚拟机，可以选择链接克隆，也可以完整克隆。

克隆完成之后，启动这三台虚拟机。

第一步：修改虚拟机 IP 地址

修改三台主机的 IP 分别为 192.168.20.10 、 192.168.20.20 、 192.168.20.30 ，当然，这个 IP 地址需要根据自己虚拟网卡的地址进行调整。

$ cd /etc/sysconfig/network-scripts/
$ vim ifcfg-ens33

文件内容如下：

TYPE=Ethernet
PROXY_METHOD=none
BROWSER_ONLY=no
BOOTPROTO=static
IPADDR=192.168.20.10
NETMASK=255.255.255.0
GATEWAY=192.168.20.2
DNS1=114.114.114.114
DEFROUTE=yes
IPV4_FAILURE_FATAL=no
NAME=ens33
UUID=b7310b0a-1fa8-4a93-b224-5b058417fe07
DEVICE=ens33
ONBOOT=yes

重启网络服务

$ systemctl restart network

第二步：修改三台机器的主机名

分别将三台主机的主机名修改为 node1 、node2 、node3 。

$ vim /etc/hostname
# 修改文件的内容为节点名称
node1
$ reboot     # 重启主机方可应用

第三步：配置各个结点的 hosts 文件

配置 hosts 文件是为了三台主机之间可以直接通过主机名互相访问。

$ vim /etc/hosts
# 三个主机的host文件都填入如下内容
192.168.20.10 node1
192.168.20.20 node2
192.168.20.30 node3

修改完后，使用 ping 命令确保连通性后，再继续后面步骤。

第四步：确保各个结点的 cookie 文件使用的是同一个值

在 node1 上执行如下远程操作命令即可

$ scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie
$ scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie

执行命令期间，如果询问是否继续，输入 yes ；要求输入密码，按要求输入密码即可。

第五步：启动 RabbitMQ 服务

在三台结点上分别输入如下命令

$ rabbitmq-server -detached

同时这条命令顺带启动 Erlang 虚拟机和 RbbitMQ 应用服务。

第六步：关闭子结点 RabbitMQ 服务，并加入集群

在 node2 中输入：

# rabbitmqctl stop 会将 Erlang 虚拟机关闭，rabbitmqctl stop_app 只关闭 RabbitMQ 服务
$ rabbitmqctl stop_app 
$ rabbitmqctl reset
$ rabbitmqctl join_cluster rabbit@node1
$ rabbitmqctl start_app        # 只启动应用服务

在 node3 中输入：

$ rabbitmqctl stop_app
$ rabbitmqctl reset
$ rabbitmqctl join_cluster rabbit@node2
$ rabbitmqctl start_app

查看集群状态

$ rabbitmqctl cluster_status

如下图所示表示成功

第七步：重新设置用户

# 创建账号
$ rabbitmqctl add_user admin 123456
# 设置用户角色
$ rabbitmqctl set_user_tags admin administrator
# 设置用户权限
$ rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

后续操作：解除集群节点

以解除 node2 节点为例子，在 node2 节点中输入：

$ rabbitmqctl stop_app
$ rabbitmqctl reset
$ rabbitmqctl start_app
$ rabbitmqctl cluster_status

最后在 node1 中输入：

$ rabbitmqctl forget_cluster_node rabbit@node2

搭建成功

如果搭建成功，访问 http://192.168.20.20:15672/ 即可进入 RabbitMQ 集群的 Web 管理界面，进入首页后应该可以看到如下三台节点。

镜像队列

介绍

默认情况下，RabbitMQ 搭建的集群之间的队列并不会共享，也就是我们在 node1 上创建的队列在 node2 中并不存在，这就会存在一个问题，如果 node1 突然宕机了，那么原本在 node1 上的所有队列也都失效了，如果要队列恢复必须等到 node1 恢复，并且如果队列没有持久化还存在数据丢失的风险。

为了避免上述的问题，就必须创建镜像队列，镜像队列可以将 node1 上的队列拷贝到另一个节点中，这样如果 node1 失效了，其他节点还可以照常提供服务，以保障服务的高可用性。

搭建方式

按照之前的方式搭建好集群后，我们可以进入任意一个节点的 Web 管理页面，然后根据下图方式，即可搭建镜像队列的策略。

策略参数说明

• Name：policy的名称。
• Pattern：queue的匹配模式(正则表达式)。
• Definitio：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode。

• ha-mode：指明镜像队列的模式，有效值为 all/exactly/nodes。

  • all：表⽰在集群中所有的节点上进⾏镜像(并不推荐)。
  • exactly：表⽰在指定个数的节点上进⾏镜像，节点的个数由ha-params指定。
  • nodes：表⽰在指定的节点上进⾏镜像，节点名称通过ha-params指定。
• ha-params：ha-mode模式需要⽤到的参数。
• ha-sync-mode：进⾏队列中消息的同步⽅式，有效值为automatic和manual。
• priority：可选参数，policy的优先级。

请注意⼀个事实，镜像配置的 pattern 采⽤的是正则表达式匹配，也就是说会匹配⼀组。

测试

在开启镜像策略后，我们添加的以 mirror 开头的队列，应该都类似如下：

这时候，当 node1 宕机，我们可以在 node1 输入 rabbitmqctl stop_app 关闭服务，那么 RabbitMQ 会自动切换到 node2 节点，并自动在 node3 节点中再镜像一份，保证队列的可用性。

注意，这时候即使 node1 节点恢复，这个队列也还是在 node2 节点中，并不会恢复到 node1 节点内。

通过镜像队列，可以保证即使集群中仅剩下一个节点了，依然能够保证队列的可用性。

Haproxy 实现负载均衡

问题

搭建完集群和镜像队列之后，还有一个问题就是我们的客户端，每次只能与其中一个节点建立连接，如下：

ConnectionFactory factory = new ConnectionFactory();
// RabbitMQ所在服务器的地址
factory.setHost("192.168.20.20");
// 端口号，默认也是5672
factory.setPort(5672);

这样即使搭建了集群，一旦某个节点宕机了，还是会导致连接中断，从而影响到应用程序，这显然也不是我们所需要的，那怎么解决这个问题呢？

我们可以在客户端和服务器之间多加一层作为网关，这个网关负责去找具体的某个 RabbitMQ 服务，而作为客户端，不需要关注具体的 RabbitMQ 服务器的地址，只需要直到网关的地址，通过网关就可以找到 RabbitMQ 服务器。

网关中可以设置多个 RabbitMQ 服务器，当客户端过来的时候，就可以将其连接到某个具体的 RabbitMQ 服务器中，从而实现负载均衡的作用。

介绍

HAProxy 提供高可用性、负载均衡及基于 TCPHTTP 应用的代理，支持虚拟主机，它是免费、快速并 且可靠的一种解决方案，包括 Twitter、Reddit、StackOverflow、GitHub 在内的多家知名互联网公司在使用。 HAProxy 实现了一种事件驱动、单一进程模型，此模型支持非常大的井发连接数。

搭建步骤

Tip：以下内容均在 node1 节点中搭建，我们也可以新建一台虚拟机，作为网关。

第一步：下载 haproxy

$ yum -y install haproxy

第二步：修改 haproxy.cfg 配置文件

$ vim /etc/haproxy/haproxy.cfg

文件内容如下：

# logging options
global
    log 127.0.0.1 local0 info
    maxconn 5120
    # haproxy 的安装地址
    chroot      /var/lib/haproxy
    pidfile     /var/run/haproxy.pid
    uid 99
    gid 99
    daemon
    quiet
    nbproc 20
defaults
    log global
    # 使用4层代理模式，”mode http”为7层代理模式
    mode tcp
    # if you set mode to tcp,then you nust change tcplog into httplog
    option tcplog
    option dontlognull
    retries 3
    option redispatch
    maxconn 2000
    contimeout 5s
     # 客户端空闲超时时间为 30秒 则HA 发起重连机制
     clitimeout 30s
     # 服务器端链接超时时间为 15秒 则HA 发起重连机制
     srvtimeout 15s 
#front-end IP for consumers and producters
listen rabbitmq_cluster
    #! 客户端访问的端口
    bind 0.0.0.0:5555
    # 配置TCP模式
    mode tcp
    # 简单的轮询
    balance roundrobin
        # rabbitmq集群节点配置 
        #inter 每隔五秒对mq集群做健康检查， 2次正确证明服务器可用，2次失败证明服务器不可用，并且配置主备机制
        server node1 192.168.20.10:5672 check inter 5000 rise 2 fall 2
        server node2 192.168.20.20:5672 check inter 5000 rise 2 fall 2
        server node3 192.168.20.30:5672 check inter 5000 rise 2 fall 2
# 配置haproxy web监控，查看统计信息
# 表示可以访问 http://192.168.20.10:8100/rabbitmq-stats 
listen stats
    # 网关的地址
    bind 192.168.20.10:8100
    mode http
    option httplog
    stats enable
    stats uri /rabbitmq-stats
    stats refresh 5s