RabbitMQ队列重发消息触发无限循环，如何有效解决该问题？

问题根因 #

你遇到的无限循环本质是maxMessagesPerPoll=-1的配置逻辑导致：该配置下单次轮询任务会持续从队列拉取消息，直到队列无可用消息才会结束本次轮询、进入20秒的等待间隔。而你使用Status.REQUEUE重发消息时，消息会被立即放回原队列头部，当前还在运行的轮询任务会直接拉取到这条刚放回的消息，进而触发无限循环。

最优实现方案：基于RabbitMQ延迟队列实现重试投递 #

该方案既可以避免单次轮询重复拉取同一条消息，还能自定义重试间隔，避免短时间内重复执行失败的处理逻辑：

1. 新增延迟等待队列，配置死信规则
   延迟队列不需要部署消费者，仅作为消息等待重试的载体，消息过期后会自动转发回原重试队列：

// 延迟队列定义：消息过期后转发到原有重试交换机
@Bean
Queue delayQueue() {
    return QueueBuilder.durable("delay-queue")
            .deadLetterExchange(ProcessQueuedMessageService.RETRY_EXCHANGE)
            .deadLetterRoutingKey(ProcessQueuedMessageService.RETRY_ROUTING_KEY)
            .build();
}

// 绑定延迟队列到重试交换机，使用独立的路由键
@Bean
Binding delayBinding() {
    return BindingBuilder.bind(delayQueue()).to(exchangeRetry()).with("delay-retry-routing-key");
}

2. 替换原有的REQUEUE重发逻辑
   处理失败需要重试时，先确认当前消息消费成功（从重试队列移除），再将消息发送到延迟队列并设置过期时间，到期后消息自动回到重试队列，等待下一轮轮询拉取：

// 1. 确认消费当前消息，从原重试队列移除
StaticMessageHeaderAccessor.getAcknowledgmentCallback(requeueMessage).acknowledge(Status.ACCEPT);
// 2. 发送到延迟队列，设置10秒后重试（可根据业务需求调整时长）
amqpTemplate.convertAndSend(exchangeRetry().getName(), "delay-retry-routing-key", requeueMessage.getPayload(), msg -> {
    msg.getMessageProperties().setExpiration("10000"); // 过期时间单位为毫秒
    // 复制原消息的所有自定义头
    requeueMessage.getHeaders().forEach((k, v) -> msg.getMessageProperties().setHeader(k, v));
    return msg;
});

3. 原有轮询配置无需修改，仍然保留maxMessagesPerPoll=-1和20秒固定延迟即可，因为重发的消息会先在延迟队列等待，不会立即回到重试队列被当前轮询拉取，自然不会触发无限循环。

可选低成本方案：调整轮询拉取数量 #

如果你暂时不想引入延迟队列，也可以通过修改轮询配置快速解决问题，适合对重试间隔要求不高的场景：
将maxMessagesPerPoll从-1调整为队列预估的最大存量消息数，比如你日常队列最多积压1000条消息就设置为1000，这样单次轮询最多拉取1000条消息就结束，后续重发的消息只会被下一轮20秒后的轮询拉取，不会出现当前轮询循环拉取的问题。
修改后的轮询配置示例：

@Bean
public IntegrationFlow inboundIntegrationFlowPaymentRetry() {
    return IntegrationFlows
            .from(Amqp.inboundPolledAdapter(connectionFactory, RetryQueue),
                    e -> e.poller(Pollers.fixedDelay(20_000).maxMessagesPerPoll(1000)).autoStartup(true))
            .handle(message -> {
                channelRequestFromQueue()
                        .send(MessageBuilder.withPayload(message.getPayload()).copyHeaders(message.getHeaders())
                                .setHeader(IntegrationConstants.QUEUED_MESSAGE, message).build());
            }).get();
}

该方案的缺点是如果队列实际积压消息超过你设置的数值，单次轮询无法拉取全部消息，需要多轮才能处理完所有积压。

注意事项 #

• 不要使用Status.REQUEUE做业务重试，该配置的设计本意是处理消费端临时故障（比如服务即将下线）的消息回滚，不是用来做业务重试的。所有业务重试都应该主动投递到队列或者延迟队列，同时要给消息增加重试次数字段，达到最大重试次数后丢入死信队列人工处理，避免消息无限重试。
• 如果需要不同消息设置不同的重试间隔，可以直接安装RabbitMQ的延迟消息插件，不需要为每个延迟时长创建独立队列，适配性更高。

内容的提问来源于stack exchange，提问作者Sanal M