Spring Integration Google Pub/Sub消息发送异常处理与配置疑问

我刚开始使用Spring Integration结合Google Pub/Sub模型向外部系统发送消息，通过Service Activator接收载荷并执行发送逻辑，核心代码如下：

@ServiceActivator(inputChannel = "inputChannel")
public void messageReceiver(final String payloadMessage) throws IOException {
    adapter.sendData(payloadMessage);  // 将数据发送到外部系统，需添加异常处理
}

需求 #

需要为adapter.sendData(payloadMessage)实现异常处理，覆盖以下场景：

• 外部系统宕机
• 我方与外部系统间的网络故障
  要求异常发生时数据不丢失，外部系统恢复后能正常接收消息。

当前配置与问题 #

已配置错误通道处理异常，但当sendData()报错时，控制台会持续重复处理同一条失败消息。相关配置代码如下：

@Bean
public PubSubInboundChannelAdapter messageChannelAdapter(final @Qualifier("myInputChannel") MessageChannel inputChannel,
        PubSubTemplate pubSubTemplate) 
{
    PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, pubSubSubscriptionName);
    adapter.setOutputChannel(inputChannel);
    adapter.setAckMode(AckMode.AUTO_ACK);
    adapter.setErrorChannelName("pubsubErrors");
    return adapter;
}

@ServiceActivator(inputChannel = "pubsubErrors")
public void pubsubErrorHandler(Message<MessagingException> exceptionMessage) {

    BasicAcknowledgeablePubsubMessage originalMessage = (BasicAcknowledgeablePubsubMessage) exceptionMessage
            .getPayload().getFailedMessage().getHeaders().get(GcpPubSubHeaders.ORIGINAL_MESSAGE);

    originalMessage.nack();
}

疑问：是否需要在yaml配置文件中添加spring.cloud.gcp.pubsub.subscriber.max-ack-extension-period参数？

问题根源 #

当前使用AckMode.AUTO_ACK模式，同时在错误通道调用nack()，会导致Pub/Sub收到nack后重复投递消息；且未配置重试次数限制、间隔策略，进而出现无限重复处理的情况。

核心调整步骤 #

1. 修改Ack模式为手动控制

将AckMode.AUTO_ACK改为AckMode.MANUAL，完全接管消息的确认逻辑，避免自动确认与手动nack的冲突：

adapter.setAckMode(AckMode.MANUAL);

2. 在业务方法中直接处理异常与确认

调整messageReceiver方法，直接捕获异常并手动控制ack/nack，无需依赖错误通道：

@ServiceActivator(inputChannel = "inputChannel")
public void messageReceiver(final Message<String> message) {
    BasicAcknowledgeablePubsubMessage originalPubSubMsg = 
        message.getHeaders().get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class);
    try {
        adapter.sendData(message.getPayload());
        originalPubSubMsg.ack(); // 发送成功，确认消息
    } catch (IOException | RuntimeException e) {
        // 捕获外部系统宕机、网络异常，触发重试
        originalPubSubMsg.nack();
    }
}

3. 配置重试策略与死信队列

仅靠nack会导致消息无限重复投递，需通过配置限制重试次数、设置间隔，并添加死信队列接收无法恢复的消息：

spring:
  cloud:
    gcp:
      pubsub:
        subscriber:
          retry:
            max-attempts: 5 # 最大重试次数
            initial-retry-delay: 1000 # 初始重试间隔（毫秒）
            retry-delay-multiplier: 2 # 重试间隔倍增系数（如1s→2s→4s...）
            max-retry-delay: 30000 # 最大重试间隔（毫秒）
        subscription:
          name: your-subscription-name
          dead-letter-topic: your-dead-letter-topic-name # 死信队列对应的Topic
          dead-letter-max-delivery-attempts: 5 # 达到该次数后转发到死信队列

4. 关于max-ack-extension-period参数的作用

该参数用于控制消息处理超时后，自动延长ack超时的最大时间。仅当sendData()处理耗时可能超过Pub/Sub默认的10秒ack超时时，才需要配置，例如：

spring.cloud.gcp.pubsub.subscriber.max-ack-extension-period: 300000 # 5分钟

它并非解决重复投递问题的核心，核心是重试策略与死信队列的配置。

内容的提问来源于stack exchange，提问作者learner-newfeatures