spring-rabbitmq-demo/
├── src/
│   └── main/
│       ├── java/
│       │   └── cn/
│       │       └── spring/
│       │           └── rabbitmq/
│       │               └── demo/
│       │                   ├── config/
│       │                   │   ├── RabbitMQConfig.java
│       │                   │   ├── RabbitmqTemplatePostProcessor.java
│       │                   │   └── SwaggerConfig.java
│       │                   ├── controller/
│       │                   │   └── rabbitmq/
│       │                   │       └── RabbitmqDemoController.java
│       │                   ├── rabbitmq/
│       │                   │   ├── consumer/
│       │                   │   │   └── MessageConsumer.java
│       │                   │   ├── message/
│       │                   │   │   ├── BaseMessage.java
│       │                   │   │   └── DemoMessage.java
│       │                   │   └── sender/
│       │                   │       └── MessageSender.java
│       │                   └── RabbitmqDemoApplication.java
│       └── resources/
│           ├── application.yaml
│           └── logback-spring.xml
└── pom.xml

<dependencies>
    <!-- Web 相關 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <!-- RabbitMQ 相關 -->
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
    </dependency>
    
    <!-- Lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    
    <!-- Swagger 相關 -->
    <dependency>
        <groupId>org.springdoc</groupId>
        <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
        <version>2.8.11</version>
    </dependency>
    
    <!-- Jackson 用于JSON序列化 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
    
    <!-- hutool 工具包 -->
    <dependency>
        <groupId>cn.hutool</groupId>
        <artifactId>hutool-all</artifactId>
        <version>5.7.15</version>
    </dependency>
</dependencies>

server:
  port: 8081
spring:
  application:
    name: spring-rabbitmq-demo
  profiles:
    active: dev

  main:
    allow-circular-references: true # 允許循環(huán)依賴
--- #############rabbitmq配置#####################
spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: xxxx
    password: xxxx
    virtual-host: /
    publisher-returns: true # 啟用發(fā)布者返回功能，如果消息沒有到達隊列，則會通知生產(chǎn)者。
    # NONE（默認值）：不返回任何信息。
    # SIMPLE（簡單模式）使用同步阻塞方式等待MQ的確認回執(zhí)，發(fā)送消息后會阻塞當前線程，直到收到確認結果，性能相對較低，因為需要等待確認結果
    # CORRELATED（關聯(lián)模式）：使用異步非阻塞方式，生產(chǎn)者發(fā)送消息后，不等待MQ的確認回執(zhí)，而是直接返回，并通過回調(diào)函數(shù)的方式通知生產(chǎn)者。
    publisher-confirm-type: NONE
    listener:
      # SIMPLE：RabbitMQ 消費者將消息分發(fā)到一個單獨的線程（Invoker Thread）進行處理，消費者線程與業(yè)務處理線程是分離的
      #特點：
          #異步處理：消息消費與業(yè)務處理在不同線程中執(zhí)行
          #更好的容錯性：業(yè)務處理異常不會直接影響 RabbitMQ 消費者線程
          #更高的資源消耗：需要額外的線程來進行消息分發(fā)和處理
          #支持并發(fā)消費者配置
      #DIRECT：適用于對延遲敏感、吞吐量要求高的場景，或者資源受限的環(huán)境
        #工作原理：監(jiān)聽器直接在 RabbitMQ 消費者線程上調(diào)用執(zhí)行，沒有額外的分發(fā)線程
          #特點：
          #同步處理：消息消費與業(yè)務處理在同一個線程中執(zhí)行
          #更低的延遲：沒有線程切換開銷
          #更少的資源消耗：不需要額外的線程池
          #更簡單的線程模型：更容易調(diào)試和分析
          #業(yè)務處理的異常會直接影響消費者線程
      #STREAM（流容器）
        #工作原理：使用 RabbitMQ Stream Client 處理消息流
          #特點：
          #專為高吞吐量和持久化流處理設計
          #支持超大規(guī)模消息保留
          #支持消費者組和消息重播
          #適用于需要處理大量歷史數(shù)據(jù)的場景
          #需要 RabbitMQ 3.9+ 版本支持
      type: simple # 使用simple類型監(jiān)聽器容器
      simple:
        # 是否在啟動時自動啟動容器，默認為true 當設置為 true 時，容器會在 Spring 應用上下文啟動完成后自動開始監(jiān)聽消息；設置為 false 時，需要手動調(diào)用 start() 方法啟動容器
        auto-startup: true
        # 偵聽器調(diào)用者線程的最小數(shù)量，默認為1 控制并發(fā)消費者的最小數(shù)量，用于處理消息的并行度
        concurrency: 1
        # 偵聽器調(diào)用者線程的最大數(shù)量，默認為1（與concurrency相同）當消息負載較高時，容器可以動態(tài)擴展到的最大消費者數(shù)量
        max-concurrency: 1
        # 每個消費者能夠同時存在且未被確認的消息的最大數(shù)量。，默認為250
        prefetch: 250
        # 確認模式，可選值：NONE（不確認）、MANUAL（手動確認）、AUTO（自動確認），默認為AUTO
        acknowledge-mode: AUTO
        # 默認情況下，拒絕交付是否重新排隊，默認為true 當監(jiān)聽器方法拋出異常時，決定消息是否重新放回隊列。設置為 false 可以避免消息無限重試
        default-requeue-rejected: true
        # 應該多久發(fā)布一次空閑容器事件，默認不發(fā)布（無默認值） 用于監(jiān)控容器狀態(tài)，當容器在指定時間內(nèi)沒有消息處理時會發(fā)布 ApplicationEvent
        idle-event-interval: 0ms
        # 是否將批處理消息作為離散消息傳遞，或者將整個批處理傳遞給監(jiān)聽器，默認為true 當啟用時，批量消息會被分解為單條消息分別處理；禁用時，整個批次作為一個消息傳遞給監(jiān)聽器
        de-batching-enabled: true
        # 當容器停止時是否立即停止還是處理完所有預取的消息，默認為false 設置為 true 時，容器會在處理完當前消息后立即停止；false 時，會處理完所有預取消息后再停止
        force-stop: false
        # 是否啟用觀察（Observation），默認為false 啟用后可以通過 Micrometer 收集容器的指標信息
        observation-enabled: false
        # 批次大小，以物理消息數(shù)量表示，默認為1 與批量處理相關，定義每個批次包含的消息數(shù)量
        batch-size: 10
        # 如果容器聲明的隊列在代理上不可用，是否失敗，默認為true 設置為 true 時，如果隊列不可用容器會失敗或停止；false 時容器會繼續(xù)運行
        missing-queues-fatal: true
        # 是否基于'receive-timeout'和'batch-size'創(chuàng)建消息批次，默認為false 啟用后會將 deBatchingEnabled 強制設為 true，將生產(chǎn)者創(chuàng)建的批次內(nèi)容作為離散記錄包含在批次中
        consumer-batch-enabled: false
        # 適中的接收超時時間
        receive-timeout: 5000ms
        # 重試相關配置
        retry:
          # 是否啟用重試機制，默認為false
          enabled: false
          # 最大嘗試次數(shù)，默認為3
          max-attempts: 3
          # 初始重試間隔時間，默認為1000ms
          initial-interval: 1000ms
          # 最大重試間隔時間，默認為10000ms
          max-interval: 10000ms
          # 重試間隔的乘數(shù)，默認為2.0
          multiplier: 2.0
          # 重試時是有狀態(tài)還是無狀態(tài)，默認為true（無狀態(tài)）
          stateless: true
    template:
      # 是否啟用強制消息投遞，默認為false
      # 當設置為true時，如果消息無法路由到隊列，會拋出AmqpMessageReturnedException異常
      # 需要配合RabbitTemplate的ReturnsCallback使用
      mandatory: false
      # receive()操作的超時時間，默認為0ms（無限等待）
      # 用于receive()方法調(diào)用時的等待超時時間
      receive-timeout: 0ms
      # sendAndReceive()操作的超時時間，默認為5000ms（5秒）
      # 用于請求-回復模式下的等待超時時間
      reply-timeout: 5000ms
      # 發(fā)送操作使用的默認交換機名稱，默認為空字符串（使用默認交換機）
      # 當使用RabbitTemplate發(fā)送消息時不指定交換機時使用此默認值
      exchange: ""
      # 發(fā)送操作使用的默認路由鍵，默認為空字符串
      # 當使用RabbitTemplate發(fā)送消息時不指定路由鍵時使用此默認值
      routing-key: ""
      # 當沒有明確指定接收隊列時，默認接收消息的隊列名稱，默認為null
      # 用于RabbitTemplate接收消息時的默認隊列
      default-receive-queue:
      # 是否啟用觀察（Observation），默認為false
      # 啟用后可以通過Micrometer收集RabbitTemplate的指標信息
      observation-enabled: false
      # 用于反序列化時允許的包/類的簡單模式列表，默認為null
      # 用于控制哪些類可以被反序列化，防止不安全的反序列化
      allowed-list-patterns:
      # 重試相關配置
      retry:
        # 是否啟用重試機制，默認為false
        # 啟用后RabbitTemplate在發(fā)送消息失敗時會進行重試
        enabled: false
        # 最大嘗試次數(shù)，默認為3次
        max-attempts: 3
        # 初始重試間隔時間，默認為1000ms（1秒）
        initial-interval: 1000ms
        # 最大重試間隔時間，默認為10000ms（10秒）
        max-interval: 10000ms
        # 重試間隔的乘數(shù)，默認為2.0
        # 每次重試的間隔時間會乘以這個數(shù)值
        multiplier: 2.0

@Data
public class BaseMessage implements java.io.Serializable {
    @Serial
    private static final long serialVersionUID = 1L;
    
    protected Long deliveryTag = IdUtil.getSnowflake().nextId();
}

@Data
public class DemoMessage extends BaseMessage{
    private Long userId;
    private String message;
    private Date createTime;
}

@Configuration
@Import(RabbitmqTemplatePostProcessor.class)
public class RabbitMQConfig {

    /**
     * 延遲交換機（用于接收延遲消息）
     *
     * @return
     */
    @Bean
    public Exchange delayExchange() {
        return ExchangeBuilder.directExchange("delay.exchange")
                .durable(true)
                .build();
    }

    /**
     * 延遲隊列（沒有消費者監(jiān)聽）
     * 設置死信交換機和路由鍵，當消息過期后自動轉(zhuǎn)發(fā)到死信交換機
     *
     * @return
     */
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue")
                // 設置死信交換機
                .deadLetterExchange("delay.process.exchange")
                // 設置死信路由鍵
                .deadLetterRoutingKey("delay.process")
                .build();
    }

    /**
     * 延遲交換機與延遲隊列綁定
     *
     * @return
     */
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue())
                .to(delayExchange())
                .with("delay")
                .noargs();
    }


    /**
     * json消息轉(zhuǎn)換器
     *
     * @return
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 為批量處理創(chuàng)建專用的監(jiān)聽器容器工廠
     *
     * @param connectionFactory
     * @return
     */
    @Bean("batchListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory batchListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(3);
        factory.setPrefetchCount(10);
        factory.setBatchListener(true); // 啟用批量監(jiān)聽
        factory.setConsumerBatchEnabled(true); // 啟用消費者端批量處理
        factory.setBatchSize(10); // 設置批次大小
        factory.setReceiveTimeout(5000L); // 設置接收超時時間
        factory.setBatchReceiveTimeout(5000L);
        factory.setDeBatchingEnabled(false); // 禁用分解批處理消息
        factory.setMessageConverter(messageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 設置為 true 可在當前消息處理完畢后停止容器，并重新排隊任何預取的消息。在使用獨占或單活動消費者時很有用
        factory.setForceStop(true);
        // 設置關閉超時時間
        factory.setContainerCustomizer((container -> {
            container.setShutdownTimeout(30000L);
            // container.setExclusive(true);
        }));
        return factory;
    }
}

publisher-returns: true # 啟用發(fā)布者返回功能，如果消息沒有到達隊列，則會通知生產(chǎn)者。
    # NONE（默認值）：不返回任何信息。
    # SIMPLE（簡單模式）使用同步阻塞方式等待MQ的確認回執(zhí)，發(fā)送消息后會阻塞當前線程，直到收到確認結果，性能相對較低，因為需要等待確認結果
    # CORRELATED（關聯(lián)模式）：使用異步非阻塞方式，生產(chǎn)者發(fā)送消息后，不等待MQ的確認回執(zhí)，而是直接返回，并通過回調(diào)函數(shù)的方式通知生產(chǎn)者。
    publisher-confirm-type: NONE

@Slf4j
public class RabbitmqTemplatePostProcessor implements BeanPostProcessor {
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof RabbitTemplate) {
            RabbitTemplate rabbitTemplate = (RabbitTemplate) bean;

            // 啟用發(fā)送確認機制（ConfirmCallback）
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    if (ack) {
                        log.info("消息成功發(fā)送到交換機，correlationData: {}", correlationData);
                    } else {
                        log.error("消息發(fā)送到交換機失敗，correlationData: {}, cause: {}", correlationData, cause);
                    }
                }
            });

            // 啟用發(fā)送失敗回調(diào)機制（ReturnCallback）
            rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
                @Override
                public void returnedMessage(ReturnedMessage returnedMessage) {
                    log.error("消息無法路由到隊列，exchange: {}, routingKey: {}, message: {}, replyCode: {}, replyText: {}",
                            returnedMessage.getExchange(),
                            returnedMessage.getRoutingKey(),
                            returnedMessage.getMessage(),
                            returnedMessage.getReplyCode(),
                            returnedMessage.getReplyText());
                }
            });

            // 設置消息持久化
            rabbitTemplate.setMandatory(true);

            log.info("RabbitTemplate 配置完成：啟用 ConfirmCallback、ReturnsCallback 和消息持久化");
        }
        return bean;
    }

}

@Component
public class MessageSender {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @Value("${rabbitmq.fanout.exchange:fanout-exchange}")
    private String fanoutExchange;

    @Value("${rabbitmq.topic.exchange:topic-exchange}")
    private String topicExchange;

    @Value("${rabbitmq.delay.exchange:delay.exchange}")
    private String delayExchange;


    /**
     * 發(fā)送消息到隊列demo.queue
     *
     * @param message
     */
    public void basicSend(DemoMessage message) {
        rabbitTemplate.convertAndSend("demo.queue", message);
    }


    /**
     * 發(fā)送消息到廣播exchange
     *
     * @param message
     */
    public void fanoutSend(DemoMessage message) {
        rabbitTemplate.convertAndSend(fanoutExchange, "", message);
    }

    /**
     * 發(fā)送消息到topic exchange
     *
     * @param message    消息內(nèi)容
     * @param routingKey 路由鍵
     */
    public void topicSend(DemoMessage message, String routingKey) {
        rabbitTemplate.convertAndSend(topicExchange, routingKey, message);
    }

    /**
     * 發(fā)送延遲消息
     *
     * @param message 消息內(nèi)容
     * @param delay   延遲時間（毫秒）
     */
    public void delaySend(DemoMessage message, long delay) {
        rabbitTemplate.convertAndSend(delayExchange, "delay", message, msg -> {
            if (delay > 0) {
                msg.getMessageProperties().setExpiration(String.valueOf(delay));
            }
            return msg;
        });
    }

    /**
     * 批量發(fā)送消息
     *
     * @param messages 消息列表
     */
    public void batchSend(List<DemoMessage> messages) {
        messages.forEach(message -> {
            rabbitTemplate.convertAndSend("batch.exchange", "batch", message, msg -> {
                // 自定義消息屬性
                return msg;
            });

        });
    }
}

@Slf4j
@Component
public class MessageConsumer {

    @Resource
    private ObjectProvider<MessageConverter> messageConverterObjectProvider;

    /**
     * 監(jiān)聽并處理DemoMessage類型的消息
     *
     * @param message 消息內(nèi)容
     */
    @RabbitListener(queuesToDeclare = {@Queue("demo.queue")})
    public void handleMessageByAnnotation(DemoMessage message) {
        log.info("[handleMessageByAnnotation] 收到消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 監(jiān)聽廣播消息1
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout.queue.1"),
            exchange = @Exchange(value = "fanout-exchange", type = "fanout")
    ))
    public void handleFanoutMessage1(DemoMessage message) {
        log.info("[handleFanoutMessage1] 收到廣播消息: userId={}, message={},  createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }


    /**
     * 監(jiān)聽廣播消息2
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout.queue.2"),
            exchange = @Exchange(value = "fanout-exchange", type = "fanout")
    ))
    public void handleFanoutMessage2(DemoMessage message) {
        log.info("[handleFanoutMessage2] 收到廣播消息: userId={}, message={},  createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 監(jiān)聽topic消息1
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue.1"),
            exchange = @Exchange(value = "topic-exchange", type = "topic"),
            key = "topic.message.specific"
    ))
    public void handleTopicMessage1(DemoMessage message) {
        log.info("[handleTopicMessage1] 收到Topic消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 監(jiān)聽topic消息2
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue.2"),
            exchange = @Exchange(value = "topic-exchange", type = "topic"),
            key = "topic.message.*"
    ))
    public void handleTopicMessage2(DemoMessage message) {
        log.info("[handleTopicMessage2] 收到Topic消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 監(jiān)聽延遲消息處理隊列
     * 當延遲隊列中的消息過期后，會被自動轉(zhuǎn)發(fā)到此隊列進行處理
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("delay.process.queue"),
            exchange = @Exchange(value = "delay.process.exchange", type = "direct"),
            key = "delay.process"
    ))
    public void handleDelayMessage(DemoMessage message) {
        log.info("[handleDelayMessage] 收到延遲消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 批量處理消息
     *
     * @param messages
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("batch.queue"),
            exchange = @Exchange(value = "batch.exchange", type = "direct"),
            key = "batch"
    ), batch = "true", containerFactory = "batchListenerContainerFactory", ackMode = "MANUAL")
    public void handleBatchMessage(List<Message> messages, Channel channel) {
        log.info("[handleBatchMessage] 開始處理批量消息，共 {} 條", messages.size());

        if (CollUtil.isEmpty(messages)) {
            log.info("[handleBatchMessage] 消息列表為空，無需處理");
            return;
        }

        // 分別存儲成功和失敗的消息
        List<Message> successMessages = new ArrayList<>();
        List<Message> failedMessages = new ArrayList<>();

        // 批量轉(zhuǎn)換消息
        for (Message message : messages) {
            try {
                DemoMessage demoMessage = (DemoMessage) messageConverterObjectProvider.getObject().fromMessage(message);
                demoMessage.setDeliveryTag(message.getMessageProperties().getDeliveryTag());
                successMessages.add(message);
                log.debug("[handleBatchMessage] 消息轉(zhuǎn)換成功: deliveryTag={}", message.getMessageProperties().getDeliveryTag());
            } catch (Exception e) {
                log.error("[handleBatchMessage] 消息轉(zhuǎn)換失敗: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                failedMessages.add(message);
            }
        }

        // 處理成功轉(zhuǎn)換的消息
        if (CollUtil.isNotEmpty(successMessages)) {
            try {
                log.info("[handleBatchMessage] 開始處理 {} 條成功轉(zhuǎn)換的消息", successMessages.size());
                // 模擬處理時間 - 實際應用中這里應該是真正的業(yè)務邏輯
                processMessages(successMessages);

                // 批量確認所有成功處理的消息
                for (Message message : successMessages) {
                    try {
                        long deliveryTag = message.getMessageProperties().getDeliveryTag();
                        channel.basicAck(deliveryTag, false);
                        log.debug("[handleBatchMessage] 消息確認成功: deliveryTag={}", deliveryTag);
                    } catch (IOException e) {
                        log.error("[handleBatchMessage] 消息確認失敗: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                    }
                }
                log.info("[handleBatchMessage] 成功處理并確認 {} 條消息", successMessages.size());

            } catch (Exception e) {
                log.error("[handleBatchMessage] 處理消息時發(fā)生異常", e);
                // 如果處理過程中出現(xiàn)異常，將所有成功轉(zhuǎn)換的消息標記為失敗
                failedMessages.addAll(successMessages);
            }
        }

        // 處理轉(zhuǎn)換失敗的消息
        if (!failedMessages.isEmpty()) {
            log.warn("[handleBatchMessage] 共 {} 條消息處理失敗，嘗試重新入隊", failedMessages.size());
            for (Message message : failedMessages) {
                try {
                    long deliveryTag = message.getMessageProperties().getDeliveryTag();
                    // 第三個參數(shù)設為true，表示將消息重新放回隊列
                    channel.basicNack(deliveryTag, false, true);
                    log.debug("[handleBatchMessage] 消息重新入隊: deliveryTag={}", deliveryTag);
                } catch (IOException e) {
                    log.error("[handleBatchMessage] 消息重新入隊失敗: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                }
            }
        }

        log.info("[handleBatchMessage] 批量消息處理完成: 成功={}條, 失敗={}條",
                successMessages.size(), failedMessages.size());
    }

    /**
     * 實際處理消息的方法
     * 在實際應用中，這里應該包含真正的業(yè)務邏輯
     *
     * @param messages 待處理的消息列表
     * @throws Exception 處理異常
     */
    private void processMessages(List<Message> messages) throws Exception {
        // 模擬處理時間
        Thread.sleep(50L);

        // 實際應用中，這里應該是真正的業(yè)務邏輯處理
        log.info("[processMessages] 處理了 {} 條消息", messages.size());
    }

@RabbitListener(..., ackMode = "MANUAL")
public void handleBatchMessage(List<Message> messages, Channel channel) {
    // 處理消息
    // 手動確認消息
    channel.basicAck(deliveryTag, false);
}

@Bean("batchListenerContainerFactory")
public SimpleRabbitListenerContainerFactory batchListenerContainerFactory(
        ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setBatchListener(true); // 啟用批量監(jiān)聽
    factory.setConsumerBatchEnabled(true); // 啟用消費者端批量處理
    factory.setBatchSize(10); // 設置批次大小
    return factory;
}

/**
     * 批量處理消息
     *
     * @param messages
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("batch.queue"),
            exchange = @Exchange(value = "batch.exchange", type = "direct"),
            key = "batch"
    ), batch = "true", containerFactory = "batchListenerContainerFactory", ackMode = "MANUAL")
    public void handleBatchMessage(List<Message> messages, Channel channel) {
    }

public void delaySend(DemoMessage message, long delay) {
    rabbitTemplate.convertAndSend(delayExchange, "delay", message, msg -> {
        if (delay > 0) {
            msg.getMessageProperties().setExpiration(String.valueOf(delay));
        }
        return msg;
    });
}


/**
     * 監(jiān)聽延遲消息處理隊列
     * 當延遲隊列中的消息過期后，會被自動轉(zhuǎn)發(fā)到此隊列進行處理
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("delay.process.queue"),
            exchange = @Exchange(value = "delay.process.exchange", type = "direct"),
            key = "delay.process"
    ))
    public void handleDelayMessage(DemoMessage message) {
    ...
    }