RabbitMQ - 4 ( 22000 字 RabbitMQ 入门级教程 )

一： RabbitMQ 高级特性

前面主要讲解了 RabbitMQ 的概念和应用。RabbitMQ 实现了 AMQP 0-9-1 规范，并在此基础上进行了多项扩展。在 RabbitMQ 官方网站中详细介绍了其特性，我们将其中一些重要且常用的特性挑选出来进行讲解。

1.1 消息确认

生产者发送的消息在到达消费者端后，可能会出现以下几种情况：

RabbitMQ 在向消费者发送消息后，会立即将该消息从队列中删除，但如果消费者处理消息时出现异常，则可能导致消息丢失。为了解决这一问题，RabbitMQ 提供了消息确认机制，用于确保消息被消费者成功接收并正确处理。在消费者订阅队列时，可以通过设置 autoAck 参数来控制消息确认机制，根据该参数的不同，消息确认机制分为两种模式。

// 创建一个 DefaultConsumer 对象，用于处理接收到的消息
DefaultConsumer consumer = new DefaultConsumer(channel) {
    // 当有消息送达时会触发该方法
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 输出接收到的消息内容，将字节数组转换为字符串
        System.out.println("接收到消息: " + new String(body));
    }
};

// 开始监听队列，指定队列名称、是否自动确认消息以及消费者对象
channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, false, consumer);

当 autoAck 参数设置为 false 时，对于 RabbitMQ 服务端而言，队列中的消息会分为两部分：一部分是尚未投递给消费者的消息，另一部分是已投递但尚未收到消费者确认的消息。如果 RabbitMQ 长时间未收到消费者的确认信号，并且该消费者已断开连接，则 RabbitMQ 会将消息重新放回队列，等待重新投递给下一个消费者，或者可能再次投递给原来的消费者。

从 RabbitMQ 的 Web 管理平台上也可以看到当前队列中 Ready 状态和 Unacked 状态的消息数

1.1.1 自动确认

自动确认之前已经提到过，这里不再赘述。

1.1.2 手动确认

消费者在收到消息后，可以选择确认消息、拒绝消息或跳过消息。RabbitMQ 提供了多种确认应答方式，消费者可以通过调用其对应的 channel 方法进行操作，主要包括以下三种方式：

下面是参数解释：

我们将基于 SpringBoot 演示消息的确认机制。与直接使用 RabbitMQ Java Client 库相比，Spring-AMQP 对消息确认机制的使用方式有所不同，但是也提供了三种策略来实现消息确认。

如果需要配置的话在配置文件中进行配置即可，下面演示一种确认模式的使用：

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.51.65:15673/bite
    listener:
      simple:
        acknowledge-mode: none

1. 生产者代码：

    public static final String ACK_EXCHANGE_NAME = "ack_exchange";
    public static final String ACK_QUEUE = "ack_queue";

@Configuration
public class RabbitMQConfig {

    // 声明交换机
    @Bean("ackExchange")
    public Exchange ackExchange() {
        return ExchangeBuilder
                .topicExchange(Constant.ACK_EXCHANGE_NAME)
                .durable(true)
                .build();
    }

    // 声明队列
    @Bean("ackQueue")
    public Queue ackQueue() {
        return QueueBuilder
                .durable(Constant.ACK_QUEUE)
                .build();
    }

    // 绑定队列和交换机
    @Bean("ackBinding")
    public Binding ackBinding(@Qualifier("ackExchange") Exchange exchange,
                               @Qualifier("ackQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("ack")
                .noargs();
    }
}

@RestController
@RequestMapping("/producer")
public class ProductController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送消息到指定交换机和路由键
    @RequestMapping("/ack")
    public String ack() {
        rabbitTemplate.convertAndSend(
                Constant.ACK_EXCHANGE_NAME, 
                "ack", 
                "consumer ack test..."
        );
        return "发送成功!";
    }
}

3. 消费者代码

@Component
public class AckQueueListener {

    // 监听队列 Constant.ACK_QUEUE
    @RabbitListener(queues = Constant.ACK_QUEUE)
    public void listenerQueue(Message message, Channel channel) throws Exception {
        try {
            // 获取消息内容和 deliveryTag
            String messageBody = new String(message.getBody(), "UTF-8");
            long deliveryTag = message.getMessageProperties().getDeliveryTag();

            System.out.printf("接收到消息: %s, deliveryTag: %d%n", messageBody, deliveryTag);

            // 模拟处理失败
            int num = 3 / 0;

            System.out.println("处理完成");
        } catch (Exception e) {
            // 处理异常逻辑，日志记录或其他操作
            System.err.println("消息处理失败: " + e.getMessage());
            // 根据业务需求，这里可以选择是否重新入队或丢弃消息
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

1.2 持久性

前面我们讨论了如何在消费端处理消息时确保消息不丢失，但如果 RabbitMQ 服务停止或崩溃后，如何确保生产者发送的消息不丢失呢？默认情况下，RabbitMQ 在退出或崩溃时会丢失队列和消息，除非明确配置其持久化机制。RabbitMQ 的持久化包括三个部分：交换机的持久化、队列的持久化和消息的持久化。

1.2.1 交换机持久化

交换器的持久化通过在声明交换机时将 durable 参数设置为 true 来实现，这会将交换机的属性保存到服务器中。当 RabbitMQ 服务器发生意外或关闭后，重启时交换机会自动恢复，无需重新创建，相当于一直存在。如果未设置持久化，则在 RabbitMQ 重启后，交换机的元数据会丢失。对于长期使用的交换机，建议将其设置为持久化，以确保其可靠性和持久性。

ExchangeBuilder.topicExchange(Constant.ACK_EXCHANGE_NAME).durable(true).build()

1.2.2 队列的持久化

队列的持久化通过在声明队列时将 durable 参数设置为 true 来实现。如果队列未设置持久化，在 RabbitMQ 服务重启后，该队列会被删除，同时其中的消息也会丢失（队列消失，消息无处存储）。队列的持久化可以保证队列本身的元数据在异常情况下不会丢失，但不能保证队列中的消息不丢失。要确保消息的可靠性，还需将消息设置为持久化。我们之前创建的队列都是持久化队列。

QueueBuilder.durable(Constant.ACK_QUEUE).build(); // 持久化队列
QueueBuilder.nonDurable(Constant.ACK_QUEUE).build(); // 非持久化队列

1.2.3 消息的持久化

要实现消息的持久化，需要 MessageProperties 中的 deliveryMode 设置为 2，即 MessageDeliveryMode.PERSISTENT。这样可以确保消息在服务器重启或出现异常时不会丢失。

public enum MessageDeliveryMode {
    NON_PERSISTENT, // 非持久化
    PERSISTENT      // 持久化
}

设置队列和消息的持久化后，RabbitMQ 服务重启后消息依旧存在。如果仅设置队列持久化，重启后消息会丢失；如果仅设置消息持久化，重启后队列消失，消息也无法存储。因此，单独设置消息持久化而不设置队列持久化是没有意义的，二者需同时设置才能确保消息可靠性。

public static final BasicProperties PERSISTENT_TEXT_PLAIN = new BasicProperties(
        "text/plain", // 内容类型 (ContentType)
        null,         // 内容编码 (ContentEncoding)
        null,         // 头信息 (Headers)
        2,            // deliveryMode: 持久化 (2 表示消息持久化)
        0,            // 优先级 (Priority)
        null,         // CorrelationId
        null,         // ReplyTo
        null,         // Expiration
        null,         // MessageId
        null,         // Timestamp
        null,         // Type
        null,         // UserId
        null,         // AppId
        null          // ClusterId
);

// 发送非持久化消息
channel.basicPublish(
        "", 
        QUEUE_NAME, 
        null, 
        msg.getBytes()
);

// 发送持久化消息
channel.basicPublish(
        "", 
        QUEUE_NAME, 
        MessageProperties.PERSISTENT_TEXT_PLAIN, // 持久化
        msg.getBytes()
);

RabbitMQ 默认会将消息视为持久化消息，除非队列被声明为非持久化或消息在发送时被标记为非持久化。但是需要注意，将所有消息都设置为持久化会显著影响 RabbitMQ 的性能，因为将消息写入磁盘的速度远远慢于写入内存的速度。对于可靠性要求不高的消息，可以选择不进行持久化处理，以提升系统的整体吞吐量。我们在选择是否持久化消息时，需要在消息的可靠性和系统吞吐量之间进行权衡。如果使用 RabbitTemplate 发送持久化消息，代码示例如下：

// 要发送的消息内容
String message = "This is a persistent message";

// 创建一个 Message 对象，并设置为持久化
MessageProperties messageProperties = new MessageProperties();
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
Message messageObject = new Message(message.getBytes(), messageProperties);

// 使用 RabbitTemplate 发送消息
rabbitTemplate.convertAndSend(
        Constant.ACK_EXCHANGE_NAME, 
        "ack", 
        messageObject
);

将交换器、队列、消息都设置了持久化之后就能百分之百保证数据不丢失了吗？答案是否定的。

1.3 发送发确认

持久化的消息在正确写入 RabbitMQ 后，还需要一段时间才能真正存入磁盘。在此过程中，RabbitMQ 并不会为每条消息都立即调用内核的 fsync 方法进行同步存盘，而是可能暂时存储在操作系统缓存中。如果此时 RabbitMQ 节点发生宕机或重启，未落盘的消息会丢失，为了解决这一问题，RabbitMQ 提供了两种机制来确保消息投递的可靠性。

1.3.1 confirm 确认模式

在 confirm 确认模式下，生产者在发送消息时可以设置一个 ConfirmCallback 监听器，无论消息是否成功到达交换机，该监听器都会被触发。如果消息成功到达交换机，则 ACK（确认标识）为 true；如果未到达，则 ACK 为 false。

1. 配置 RabbitMQ

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.51.65:15673/bite
    listener:
      simple:
        acknowledge-mode: manual # 消息接收确认
    publisher-confirm-type: correlated # 消息发送确认

2. 设置确认回调逻辑并发送消息：无论消息确认成功还是失败，都会触发 ConfirmCallback 的 confirm 方法。如果消息成功发送到 Broker，则 ack 为 true；如果消息发送失败，则 ack 为 false，同时 cause 提供失败的原因。

@Configuration
public class RabbitTemplateConfig {

    // 配置带 ConfirmCallback 的 RabbitTemplate
    @Bean("confirmRabbitTemplate")
    public RabbitTemplate confirmRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 设置 ConfirmCallback 回调函数，通过 Lambda 表达式实现 ConfirmCallback 接口的 confirm 方法。
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.printf("消息接收成功, id: %s%n", correlationData != null ? correlationData.getId() : "null");
            } else {
                System.out.printf("消息接收失败, id: %s, cause: %s%n",
                        correlationData != null ? correlationData.getId() : "null", cause);
            }
        });

        return rabbitTemplate;
    }
}

@RestController
@RequestMapping("/producer")
public class ConfirmController {

    @Resource(name = "confirmRabbitTemplate")
    private RabbitTemplate confirmRabbitTemplate;

    // 发送消息并带有确认机制
    @RequestMapping("/confirm")
    public String confirm() {
        CorrelationData correlationData = new CorrelationData("1");

        // 发送消息
        confirmRabbitTemplate.convertAndSend(
                Constant.CONFIRM_EXCHANGE_NAME, 
                "confirm", 
                "confirm test...", 
                correlationData
        );

        return "消息发送确认完成";
    }
}

1.3.2 return 退回模式

在 RabbitMQ 的 return 退回模式中，消息到达 Exchange 后会根据路由规则匹配将消息投递到队列。如果消息无法被任何队列消费，可以选择将消息退回给发送者。此时，可以通过设置一个返回回调方法，对被退回的消息进行处理。

1. 配置 RabbitMQ

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.51.65:15673/bite
    listener:
      simple:
        acknowledge-mode: manual # 消息接收确认
    publisher-confirm-type: correlated # 消息发送确认

2. 设置返回回调逻辑并发送消息，当消息无法被路由到任何队列时会被退回给发送者，此时触发通过 setReturnCallback 设置的回调方法进行处理。

@Bean("confirmRabbitTemplate")
public RabbitTemplate confirmRabbitTemplate(CachingConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

    // 设置强制退回机制
    rabbitTemplate.setMandatory(true);

    // 设置 ReturnsCallback 回调方法
    rabbitTemplate.setReturnsCallback(returned -> 
        System.out.printf("消息被退回: %s%n", returned)
    );

    return rabbitTemplate;
}

@RestController
@RequestMapping("/producer")
public class MessageController {

    @Resource(name = "confirmRabbitTemplate")
    private RabbitTemplate confirmRabbitTemplate;

    // 测试消息退回
    @RequestMapping("/msgReturn")
    public String msgReturn() {
        // 创建 CorrelationData 对象
        CorrelationData correlationData = new CorrelationData("2");

        // 发送消息
        confirmRabbitTemplate.convertAndSend(
                Constants.CONFIRM_EXCHANGE, 
                "confirm11", 
                "message return test...", 
                correlationData
        );

        return "消息发送成功";
    }
}

1.4 总结一下 RabbitMQ 如何保证消息的可靠传输

先展示一张 RabbitMQ 消息传递的示意图：

1.5 重试机制

在消息传递过程中，可能会因网络故障、服务不可用或资源不足等问题导致消息处理失败。为了解决这些问题，RabbitMQ 提供了重试机制，允许消息在处理失败后重新发送，如果对异常进⾏捕获那么就不会进⾏重试。然而，如果失败是由程序逻辑错误引起的，多次重试也无济于事，此时可以通过设置重试次数来避免无限重试的情况。

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.51.65:15673/bite
    listener:
      simple:
        acknowledge-mode: auto # 消息接收确认
        retry:
          enabled: true          # 开启消费者失败重试
          initial-interval: 5000 # 初始失败等待时长为5秒
          max-attempts: 5        # 最大重试次数（包括首次消费）

在手动确认模式下，重试次数的限制不会像自动确认模式那样直接生效，因为是否重试以及何时重试更多取决于应用程序的逻辑和消费者的实现。在自动确认模式下，RabbitMQ 会在消息投递给消费者后自动确认消息。如果消费者在处理消息时抛出异常，RabbitMQ 会根据配置的重试参数自动将消息重新入队，从而实现重试。重试次数和重试间隔等参数可以直接在 RabbitMQ 的配置中设定，RabbitMQ 会负责执行这些重试策略。

而在手动确认模式下，消费者需要显式地对消息进行确认。如果消费者在处理消息时遇到异常，可以选择不确认消息，让消息重新入队。此时，重试的控制权完全由应用程序掌握，而不是依赖 RabbitMQ 的内部机制。应用程序可以通过自身逻辑和结合 RabbitMQ 的高级特性，灵活实现自定义的重试策略。

1.6 TTL

TTL（Time to Live，过期时间）是 RabbitMQ 提供的一种机制，可以为消息或队列设置存活时间。当消息超过指定的存活时间且未被消费时，会被自动清除。这一机制常用于一些特定场景，例如在电商平台中，下单后超过24小时未付款，订单会自动取消；或者申请退款后超过7天未处理，则系统会自动退款。

目前有两种方式可以设置消息的 TTL：一种是为队列设置 TTL，此时队列中的所有消息共享相同的过期时间；另一种是为每条消息单独设置 TTL，使每条消息的过期时间可以不同。如果同时使用这两种方式，消息的实际 TTL 以两者中较小的数值为准。

1.6.1 设置消息的TTL

@RestController
@RequestMapping("/producer")
public class TtlController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送消息并设置 TTL
    @RequestMapping("/ttl")
    public String sendTtlMessage() {
        String ttlTime = "10000"; // 10秒

        // 发送消息并设置过期时间
        rabbitTemplate.convertAndSend(
                Constant.TTL_EXCHANGE_NAME, 
                "", 
                "ttl test...", 
                messagePostProcessor -> {
                	// messagePostProcessor 是一个消息后处理器，MessageProperties 包含了消息的元数据信息，例如内容类型、优先级、消息ID以及过期时间等。setExpiration(ttlTime) 是 MessageProperties 类中的一个方法，用于设置消息的过期时间。
                    messagePostProcessor.getMessageProperties().setExpiration(ttlTime);
                    return messagePostProcessor;
                }
        );

        return "发送成功!";
    }
}

1.6.2设置队列的TTL：

设置队列 TTL 可以在创建队列时通过添加参数 x-message-ttl 实现，单位为毫秒。

@Configuration
public class TtlQueueConfig {

    // 声明队列 ttlQueue2，设置过期时间为 20 秒
    @Bean("ttlQueue2")
    public Queue ttlQueue2() {
        return QueueBuilder
                .durable(Constant.TTL_QUEUE2) // 设置队列持久化
                .ttl(20 * 1000)               // 设置 TTL 为 20 秒
                .build();
    }

    // 或者通过参数设置过期时间为 20 秒
    @Bean("ttlQueue2Alt")
    public Queue ttlQueue2Alt() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 20000); // 20 秒过期
        return QueueBuilder
                .durable(Constant.TTL_QUEUE2)
                .withArguments(arguments)
                .build();
    }

    // 队列和交换机绑定
    @Bean("ttlBinding2")
    public Binding ttlBinding2(@Qualifier("ttlExchange") FanoutExchange exchange,
                                @Qualifier("ttlQueue2") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange);
    }
}

运行后会发现新增了一个队列，该队列的 Features 中显示有一个 TTL 标识，表示已成功设置消息的过期时间。

1.6.3 两种方式的区别

1.7 死信队列

死信是指因各种原因无法被正常消费的消息。当消息在一个队列中变成死信后，会被重新发送到一个专门的交换机，称为死信交换机（Dead Letter Exchange，DLX）。绑定到死信交换机的队列被称为死信队列（Dead Letter Queue，DLQ）。消息通常因以下原因变成死信：

死信交换机和死信队列本质上与普通的交换机和队列没有区别，它们只是被专门用来处理无法被正常消费的消息。

//死信队列
public static final String DLX_EXCHANGE_NAME = "dlx_exchange";
public static final String DLX_QUEUE = "dlx_queue";

@Configuration
public class DLXConfig {

    // 声明死信交换机
    @Bean("dlxExchange")
    public Exchange dlxExchange() {
        return ExchangeBuilder
                .topicExchange(Constant.DLX_EXCHANGE_NAME)
                .durable(true) // 设置为持久化
                .build();
    }

    // 声明死信队列
    @Bean("dlxQueue")
    public Queue dlxQueue() {
        return QueueBuilder
                .durable(Constant.DLX_QUEUE) // 设置为持久化
                .build();
    }

    // 绑定死信队列和死信交换机
    @Bean("dlxBinding")
    public Binding dlxBinding(@Qualifier("dlxExchange") Exchange exchange,
                               @Qualifier("dlxQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("dlx") // 设置路由键
                .noargs();
    }
}

上述代码只是声明了一个队列和交换机，此时他们还不是死信的，我们需要通过下面的代码指定死信队列和死信交换机：

@Bean("normalQueue")
public Queue normalQueue() {
    // 设置队列参数
    Map<String, Object> arguments = new HashMap<>();
    // 这两行代码指定哪些队列和交换机是死信的
    arguments.put("x-dead-letter-exchange", Constant.DLX_EXCHANGE_NAME); // 绑定死信交换机
    arguments.put("x-dead-letter-routing-key", "dlx"); // 设置死信队列的路由键

    // 创建普通队列并设置参数
    return QueueBuilder
            .durable(Constant.NORMAL_QUEUE) // 设置队列持久化
            .withArguments(arguments)       // 添加参数
            .build();
}

死信队列和死信交换机的指定可以简写为：

return QueueBuilder
        .durable(Constant.NORMAL_QUEUE)          // 声明普通队列并设置持久化
        .deadLetterExchange(Constant.DLX_EXCHANGE_NAME) // 设置死信交换机
        .deadLetterRoutingKey("dlx")             // 设置死信队列的路由键
        .build();                                // 构建队列

队列 Features 说明：

对于 RabbitMQ 来说，死信队列是一个非常有用的特性。它可以在消息无法被消费者正常消费时，将这些消息放入死信队列中进行处理。应用程序可以通过消费死信队列中的消息，分析异常原因，从而改善和优化系统的稳定性和可靠性。

在支付场景中，当用户完成支付后，支付系统会将支付结果发送到订单系统。为了避免在消息传递或处理过程中因异常导致支付信息丢失，可以使用死信队列机制。当消息处理失败时，异常消息会被投递到死信队列。订单系统的其他消费者可以监听死信队列，对这些消息进行进一步处理，例如生成工单进行人工核实或执行补偿操作，确保支付信息的完整性和系统的可靠性，死信队列还有别的应用场景：

1.8 延迟队列

延迟队列（Delayed Queue）是一种消息队列机制，消息在发送后不会立即被消费者获取，而是经过指定的延迟时间后才可供消费者消费，延迟队列有广泛的应用场景，例如：

RabbitMQ 本身不直接支持延迟队列功能，通常可以通过以下两种方式实现：

通过结合 TTL 和死信队列的方式可以实现模拟延迟队列的功能。例如在一个应用中，需要让每条消息延迟 10 秒后被消费。生产者将消息通过 normal_exchange 发送到 normal_queue 队列，并为该队列设置消息的 TTL 为 10 秒。当消息在 normal_queue 中过期后，会被转移到绑定了死信交换机的 dlx_queue 队列。消费者订阅 dlx_queue 队列，从而在延迟 10 秒后消费到这条消息。

注意事项：当发送 20 秒过期的消息后，再发送 10 秒过期的消息，会发现 10 秒过期的消息也是在 20 秒后才进入死信队列。这是因为 RabbitMQ 只会检查队首消息是否过期，如果队首消息未过期，即使后续消息已达到过期时间，也不会被立即丢弃并转移到死信队列。这会导致第一个消息的延时时间较长时，第二个消息的延时时间即使较短，也无法优先执行。

因此，在使用 TTL + 死信队列实现延迟任务队列时，需要确保业务中每个任务的延迟时间是一致的。如果业务需要为不同的任务类型设置不同的延迟时间，则需要为每种延迟时间分别创建单独的消息队列，以避免消息处理顺序受到影响。

1.8.1 延迟队列插件

RabbitMQ 官方提供了一个延迟插件来实现延迟功能，详情可参考 RabbitMQ 官方文档。安装插件后，可以通过 RabbitMQ 管理平台验证是否成功：在新建交换机时，查看是否出现延迟消息选项（Delayed Message）。如果该选项可用，说明延迟消息插件已成功运行。

1.8.2 基于插件延迟队列实现延迟队列

1. 声明交换机、队列和绑定关系

@Configuration
public class DelayedConfig {

    // 声明延迟交换机
    @Bean("delayedExchange")
    public Exchange delayedExchange() {
        return ExchangeBuilder
                .directExchange(Constant.DELAYED_EXCHANGE_NAME)
                .durable(true)   // 设置为持久化
                .delayed()       // 开启延迟功能
                .build();
    }

    // 声明队列
    @Bean("delayedQueue")
    public Queue delayedQueue() {
        return QueueBuilder
                .durable(Constant.DELAYED_QUEUE) // 设置为持久化
                .build();
    }

    // 绑定队列和延迟交换机
    @Bean("delayedBinding")
    public Binding delayedBinding(@Qualifier("delayedExchange") Exchange exchange,
                                   @Qualifier("delayedQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("delayed") // 设置路由键
                .noargs();
    }
}

2. 编写生产者代码

@RestController
@RequestMapping("/producer")
public class DelayedMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送带延迟的消息
    @RequestMapping("/delay2")
    public String sendDelayedMessages() {
        // 发送 20 秒延迟的消息
        rabbitTemplate.convertAndSend(
                Constant.DELAYED_EXCHANGE_NAME,
                "delayed",
                "delayed test 20s... " + new Date(),
                messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setDelay(20000); // 设置 20 秒延迟
                    return messagePostProcessor;
                }
        );

        // 发送 10 秒延迟的消息
        rabbitTemplate.convertAndSend(
                Constant.DELAYED_EXCHANGE_NAME,
                "delayed",
                "delayed test 10s... " + new Date(),
                messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setDelay(10000); // 设置 10 秒延迟
                    return messagePostProcessor;
                }
        );

        return "发送成功!";
    }
}

3. 编写消费者代码

@Component
public class DelayedQueueListener {

    // 监听延迟队列的消息
    @RabbitListener(queues = Constant.DELAYED_QUEUE)
    public void listenerDLXQueue(Message message, Channel channel) throws Exception {
        // 打印接收到的消息和当前时间
        System.out.printf("%tc 死信队列接收到消息: %s%n", 
                          new Date(), 
                          new String(message.getBody(), "UTF-8"));
    }
}

1.9 事务

RabbitMQ 基于 AMQP 协议实现，该协议支持事务机制，因此 RabbitMQ 也具备事务功能。同时，Spring AMQP 提供了对事务操作的支持。通过 RabbitMQ 的事务机制，开发者可以确保消息的发送和接收具有原子性，即要么全部成功，要么全部失败。

1. 配置事务管理器

@Configuration
public class TransactionConfig {

    // 配置 RabbitMQ 事务管理器
    @Bean
    public RabbitTransactionManager transactionManager(CachingConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }

    // 配置支持事务的 RabbitTemplate
    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setChannelTransacted(true); // 启用事务支持
        return rabbitTemplate;
    }
}

2. 声明队列

@Bean("transQueue")
public Queue transQueue() {
    return QueueBuilder
            .durable("trans_queue") // 声明持久化队列
            .build();
}

3. 编写生产者代码

@RestController
@RequestMapping("/trans")
public class TransactionProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送事务消息
    @Transactional
    @RequestMapping("/send")
    public String send() {
        // 发送第一条消息
        rabbitTemplate.convertAndSend("", "trans_queue", "trans test 1...");

        // 模拟异常
        int a = 5 / 0;

        // 发送第二条消息
        rabbitTemplate.convertAndSend("", "trans_queue", "trans test 2...");

        return "发送成功";
    }
}

如果不添加 @Transactional，即使代码中发生异常，消息1仍然会成功发送，当添加了 @Transactional 后，发生异常时，事务将回滚，导致消息1和消息2都不会发送成功。

1.10 消息分发

当 RabbitMQ 队列有多个消费者时，消息会被分发给不同的消费者，每条消息只会发送给一个订阅了该队列的消费者。这种方式非常适合扩展处理能力，当负载加重时，可以通过增加更多的消费者来分担消息处理任务。然而，默认情况下，RabbitMQ 以轮询方式分发消息，而不考虑消费者是否已处理并确认消息。这可能会导致部分消费者处理速度慢、消息堆积，而其他消费者处理速度快却处于空闲状态，从而降低系统整体的吞吐量。

为了解决这一问题，可以使用 channel.basicQos(int prefetchCount) 方法来限制每个消费者未确认消息的最大数量。例如，当消费者调用了 channel.basicQos(5) 后，RabbitMQ 会为该消费者计数，每发送一条消息计数加1，每确认一条消息计数减1。当未确认消息数量达到5时，RabbitMQ 将停止向该消费者发送新消息，直到其确认了部分消息。这种机制类似于 TCP/IP 中的滑动窗口，能够有效均衡消费者的负载，提高系统效率，消息分发的常见应用场景如下:

1.10.1 限流

在订单系统中，订单系统每秒最多可处理 5000 个请求，正常情况下能够满足需求，但在秒杀高峰期，请求量瞬间激增到每秒 1 万个，如果这些请求全部通过 MQ 发送到订单系统，可能会导致系统崩溃。为了解决这一问题，RabbitMQ 提供了限流机制，通过设置 prefetchCount 参数，控制消费者每次从队列中预取的消息数量，从而实现流量控制和负载均衡。同时，限流机制要求将消息应答方式设置为手动应答，以确保消费者处理完消息后再拉取新的消息。

1. 配置 prefetch 参数, 设置应答方式为⼿动应答：

# ack 确认方式：开启手动 ack
listener:
  simple:
    acknowledge-mode: manual # 手动确认
    prefetch: 5             # 每次预取 5 条消息

2. 配置交换机和队列

@Configuration
public class QosConfig {

    // 声明限流交换机
    @Bean("qosExchange")
    public Exchange qosExchange() {
        return ExchangeBuilder
                .directExchange(Constant.QOS_EXCHANGE_NAME) // 直连交换机
                .durable(true)                               // 持久化
                .build();
    }

    // 声明队列
    @Bean("qosQueue")
    public Queue qosQueue() {
        return QueueBuilder
                .durable(Constant.QOS_QUEUE) // 持久化队列
                .build();
    }

    // 绑定队列和交换机
    @Bean("qosBinding")
    public Binding qosBinding(@Qualifier("qosExchange") Exchange exchange,
                               @Qualifier("qosQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("qos") // 路由键
                .noargs();
    }
}

3. 发送消息, ⼀次发送 20 条消息

@RestController
@RequestMapping("/qos")
public class QosController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 发送消息
    @RequestMapping("/send")
    public String sendQosMessages() {
        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(
                    Constant.QOS_EXCHANGE_NAME, 
                    "qos", 
                    "qos test..." + i
            );
        }
        return "发送成功!";
    }
}

4. 编写消费者代码

@Component
public class QosQueueListener {

    // 监听指定队列的消息
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQueue(Message message, Channel channel) throws Exception {
        // 获取消息的 deliveryTag
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 打印接收到的消息和 deliveryTag
        System.out.printf("接收到消息: %s, deliveryTag: %d%n", 
                          new String(message.getBody(), "UTF-8"), 
                          deliveryTag);

        // 手动签收（可启用）
        channel.basicAck(deliveryTag, true);
    }
}

1.10.2 负载均衡

我们可以通过设置 prefetch=1 的方式，实现消息的负载均衡。比如，在有两个消费者的情况下，如果一个消费者处理任务非常快，而另一个处理较慢，默认情况下，RabbitMQ 只在消息进入队列时分发消息，不考虑消费者未确认消息的数量，可能会导致一个消费者一直繁忙，而另一个消费者空闲。通过设置 prefetch=1，可以让 RabbitMQ 一次只分配一条消息给消费者，直到该消费者处理并确认当前消息后，才会发送下一条消息。如果某个消费者繁忙，RabbitMQ 会将消息分派给下一个空闲的消费者，从而更好地实现负载均衡。

1. 配置 prefetch 参数, 设置应答方式为⼿动应答

# 配置 ack 确认方式：开启手动 ack 和预取限制
listener:
  simple:
    acknowledge-mode: manual # 手动确认
    prefetch: 1             # 每次预取 1 条消息

@Component
public class QosQueueListener {

    // 指定监听队列的名称，消费者1
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQosQueue(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 打印接收到的消息
        System.out.printf("接收到消息: %s, deliveryTag: %d%n",
                          new String(message.getBody(), "UTF-8"), deliveryTag);

        // 手动签收
        channel.basicAck(deliveryTag, true);
    }

    // 指定监听队列的名称，消费者2
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQueue2(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 打印接收到的消息
        System.out.printf("消费者2接收到消息: %s, deliveryTag: %d%n",
                          new String(message.getBody(), "UTF-8"), deliveryTag);

        // 模拟处理流程慢
        Thread.sleep(100);

        // 手动签收
        channel.basicAck(deliveryTag, true);
    }
}

接收到消息: qos test...1, deliveryTag: 1
消费者2接收到消息: qos test...0, deliveryTag: 1
接收到消息: qos test...2, deliveryTag: 2
接收到消息: qos test...3, deliveryTag: 3
接收到消息: qos test...4, deliveryTag: 4
接收到消息: qos test...5, deliveryTag: 5
消费者2接收到消息: qos test...6, deliveryTag: 2
接收到消息: qos test...7, deliveryTag: 6
接收到消息: qos test...8, deliveryTag: 7
接收到消息: qos test...9, deliveryTag: 8
接收到消息: qos test...10, deliveryTag: 9
消费者2接收到消息: qos test...11, deliveryTag: 3
接收到消息: qos test...12, deliveryTag: 10
接收到消息: qos test...13, deliveryTag: 11
接收到消息: qos test...14, deliveryTag: 12
接收到消息: qos test...15, deliveryTag: 13
消费者2接收到消息: qos test...16, deliveryTag: 4
接收到消息: qos test...17, deliveryTag: 14
接收到消息: qos test...18, deliveryTag: 15
接收到消息: qos test...19, deliveryTag: 16

deliveryTag 出现重复是因为两个消费者使用的是不同的 Channel，每个 Channel 上的 deliveryTag 都是独立计数的，因此在不同的 Channel 中可能会出现相同的 deliveryTag 值。