Spring Cloud GCP PubSub Binder轮询maxFetchSize不生效问题求助

问题描述 #

使用Spring Cloud GCP Binder从GCP PubSub主题轮询消息时，配置spring.cloud.stream.gcp.pubsub.default.consumer.maxFetchSize期望每次轮询获取指定数量的消息，但无论将该值设为2或其他数值，即使主题中存在多条消息，每次轮询仍仅获取1条。

环境信息 #

• Spring Cloud版本：2021.0.3
• Spring Cloud GCP依赖版本：3.3.0
• Java版本：OpenJDK 11

配置与代码 #

application.properties #

spring.cloud.gcp.pubsub.project-id= 
spring.cloud.gcp.credentials.location= 
spring.cloud.stream.default-binder=pubsub
spring.cloud.stream.bindings.input.destination=iamatopic
spring.cloud.stream.bindings.input.content-type=text/plain;charset=UTF-8
spring.cloud.stream.gcp.pubsub.default.consumer.maxFetchSize=3

PollableSink接口 #

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.binder.PollableMessageSource;

public interface PollableSink {
    @Input("input")
    PollableMessageSource input();
}

轮询配置类 #

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.annotation.Scheduled;

import com.google.cloud.spring.pubsub.support.AcknowledgeablePubsubMessage;
import com.google.cloud.spring.pubsub.support.GcpPubSubHeaders;

import lombok.extern.slf4j.Slf4j;

@Configuration
@EnableBinding(PollableSink.class)
@Slf4j
public class PollConfiguration {

    @Autowired
    PollableMessageSource destIn;

    PolledMessageHandler messageHandler = new PolledMessageHandler();

    @Scheduled(fixedRate = 5000)
    public void poller() {
        log.info("start polling ");
        destIn.poll(this.messageHandler, ParameterizedTypeReference.forType(String.class));
        log.info("end polling ");
    }

    static class PolledMessageHandler implements MessageHandler {
        @Override
        public void handleMessage(Message<?> message) {
            AcknowledgeablePubsubMessage ackableMessage = (AcknowledgeablePubsubMessage) message.getHeaders()
                    .get(GcpPubSubHeaders.ORIGINAL_MESSAGE);
            ackableMessage.ack();

            System.out.println("get payload : " + message.getPayload());
        }
    }
}

解决思路 #

1. 修正轮询逻辑：循环调用poll方法 #

PollableMessageSource.poll()的单次调用仅会处理一条消息，即使底层通过maxFetchSize拉取了多条消息。要批量处理所有拉取到的消息，需要在轮询任务中循环调用poll，直到返回false（表示当前批次没有更多消息）。

修改后的轮询方法：

@Scheduled(fixedRate = 5000)
public void poller() {
    log.info("start polling ");
    boolean hasMoreMessages;
    do {
        hasMoreMessages = destIn.poll(this.messageHandler, ParameterizedTypeReference.forType(String.class));
    } while (hasMoreMessages);
    log.info("end polling ");
}

2. 确认配置的优先级与正确性 #

全局默认配置spring.cloud.stream.gcp.pubsub.default.consumer.maxFetchSize对所有消费者生效，但如果需要针对特定binding单独配置，应使用binding级别的配置（优先级更高）：

spring.cloud.stream.bindings.input.consumer.maxFetchSize=3

3. 验证PubSub客户端拉取逻辑 #

maxFetchSize对应PubSub客户端的maxMessages参数，控制单次拉取的消息数量。在轮询模式下，客户端会一次性拉取配置数量的消息到本地缓存，然后通过多次poll调用逐个返回。因此循环调用poll才能消费完所有拉取到的消息。

4. 检查消息确认逻辑 #

当前代码中处理完每条消息后立即调用ack()是正确的，不会影响后续消息的拉取。但要确保没有因异常导致ack失败，进而影响订阅的消息可见性。

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