Spring Cloud Stream下GCP Pub/Sub手动ACK优化与模式判断咨询

Spring Cloud Stream提供了通用的Acknowledgment接口，无需依赖GCP特定的BasicAcknowledgeablePubsubMessage就能实现手动消息确认，完全和绑定器解耦。

你可以直接在消费方法中注入Acknowledgment参数，或者通过AcknowledgmentHeaders.ACKNOWLEDGMENT这个通用消息头获取：

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.cloud.stream.binder.Acknowledgment;
import org.springframework.cloud.stream.binder.AcknowledgmentHeaders;

@StreamListener(Sink.INPUT)
public void processMessage(String payload, @Header(AcknowledgmentHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
    // 执行业务逻辑
    // 手动确认消息
    acknowledgment.acknowledge();
    // 如果需要拒绝消息（触发重试或死信队列，取决于绑定器配置）
    // acknowledgment.nack();
}

只要消费者的ack-mode配置为MANUAL，Spring Cloud Stream就会自动传递这个Acknowledgment对象，无需任何GCP专属代码。

你可以通过Spring Cloud Stream的BindingServiceProperties来读取绑定的实际配置，从而判断当前的确认模式：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.binder.ConsumerProperties;

@Component
public class AckModeChecker {

    private final BindingServiceProperties bindingServiceProperties;

    @Autowired
    public AckModeChecker(BindingServiceProperties bindingServiceProperties) {
        this.bindingServiceProperties = bindingServiceProperties;
    }

    public boolean isManualAcknowledgment(String bindingName) {
        ConsumerProperties consumerProps = bindingServiceProperties.getBindings().get(bindingName).getConsumer();
        return "MANUAL".equalsIgnoreCase(consumerProps.getAckMode());
    }
}

调用isManualAcknowledgment方法时，传入你的消费者绑定名称（比如配置文件中spring.cloud.stream.bindings.input.consumer对应的input），就能判断当前是否为手动确认模式。

另外，也可以在消费方法中通过Acknowledgment参数是否为空快速判断：如果参数不为null，说明当前是MANUAL模式；如果为null，则大概率是AUTO模式（不过这种方式依赖绑定器实现，通过配置读取的方式更可靠）。

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