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

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

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

    /**
     * 延遲交換機(jī)與延遲隊(duì)列綁定
     *
     * @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); // 啟用消費(fèi)者端批量處理
        factory.setBatchSize(10); // 設(shè)置批次大小
        factory.setReceiveTimeout(5000L); // 設(shè)置接收超時(shí)時(shí)間
        factory.setBatchReceiveTimeout(5000L);
        factory.setDeBatchingEnabled(false); // 禁用分解批處理消息
        factory.setMessageConverter(messageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 設(shè)置為 true 可在當(dāng)前消息處理完畢后停止容器，并重新排隊(duì)任何預(yù)取的消息。在使用獨(dú)占或單活動(dòng)消費(fèi)者時(shí)很有用
        factory.setForceStop(true);
        // 設(shè)置關(guān)閉超時(shí)時(shí)間
        factory.setContainerCustomizer((container -> {
            container.setShutdownTimeout(30000L);
            // container.setExclusive(true);
        }));
        return factory;
    }
}

publisher-returns: true # 啟用發(fā)布者返回功能，如果消息沒有到達(dá)隊(duì)列，則會(huì)通知生產(chǎn)者。
    # NONE（默認(rèn)值）：不返回任何信息。
    # SIMPLE（簡(jiǎn)單模式）使用同步阻塞方式等待MQ的確認(rèn)回執(zhí)，發(fā)送消息后會(huì)阻塞當(dāng)前線程，直到收到確認(rèn)結(jié)果，性能相對(duì)較低，因?yàn)樾枰却_認(rèn)結(jié)果
    # CORRELATED（關(guān)聯(lián)模式）：使用異步非阻塞方式，生產(chǎn)者發(fā)送消息后，不等待MQ的確認(rèn)回執(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ā)送確認(rèn)機(jī)制（ConfirmCallback）
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    if (ack) {
                        log.info("消息成功發(fā)送到交換機(jī)，correlationData: {}", correlationData);
                    } else {
                        log.error("消息發(fā)送到交換機(jī)失敗，correlationData: {}, cause: {}", correlationData, cause);
                    }
                }
            });

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

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

            log.info("RabbitTemplate 配置完成：?jiǎn)⒂?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ā)送消息到隊(duì)列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   延遲時(shí)間（毫秒）
     */
    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)聽延遲消息處理隊(duì)列
     * 當(dāng)延遲隊(duì)列中的消息過期后，會(huì)被自動(dòng)轉(zhuǎn)發(fā)到此隊(duì)列進(jìn)行處理
     *
     * @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;
        }

        // 分別存儲(chǔ)成功和失敗的消息
        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());
                // 模擬處理時(shí)間 - 實(shí)際應(yīng)用中這里應(yīng)該是真正的業(yè)務(wù)邏輯
                processMessages(successMessages);

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

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

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

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

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

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

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

@Bean("batchListenerContainerFactory")
public SimpleRabbitListenerContainerFactory batchListenerContainerFactory(
        ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setBatchListener(true); // 啟用批量監(jiān)聽
    factory.setConsumerBatchEnabled(true); // 啟用消費(fèi)者端批量處理
    factory.setBatchSize(10); // 設(shè)置批次大小
    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)聽延遲消息處理隊(duì)列
     * 當(dāng)延遲隊(duì)列中的消息過期后，會(huì)被自動(dòng)轉(zhuǎn)發(fā)到此隊(duì)列進(jìn)行處理
     *
     * @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) {
    ...
    }