SpringBoot中PubSubReactiveFactory使用咨询及内存溢出问题求助

依赖配置 #

首先确保项目引入Spring Cloud GCP Pub/Sub starter和WebFlux（提供Reactor支持），Maven配置示例：

<dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

配置PubSubReactiveFactory Bean #

创建配置类，通过注入PubSubTemplate实例化PubSubReactiveFactory：

import com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.reactive.PubSubReactiveFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class PubSubReactiveConfig {

    @Bean
    public PubSubReactiveFactory pubSubReactiveFactory(PubSubTemplate pubSubTemplate) {
        return new PubSubReactiveFactory(pubSubTemplate);
    }
}

实现带背压的消息消费 #

编写消费服务类，用PubSubReactiveFactory替代传统ChannelAdapter，创建支持背压的消息流：

import com.google.cloud.spring.pubsub.reactive.PubSubReactiveFactory;
import com.google.cloud.spring.pubsub.support.AcknowledgeablePubsubMessage;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import javax.annotation.PostConstruct;

@Service
public class PubSubReactiveConsumer {

    private final PubSubReactiveFactory reactiveFactory;
    // 替换为你的订阅名称
    private static final String SUBSCRIPTION_NAME = "your-subscription-id";

    public PubSubReactiveConsumer(PubSubReactiveFactory reactiveFactory) {
        this.reactiveFactory = reactiveFactory;
    }

    @PostConstruct
    public void startConsuming() {
        // 创建支持背压的消息流
        Flux<AcknowledgeablePubsubMessage> messageFlux = reactiveFactory
                .subscribe(SUBSCRIPTION_NAME)
                // 配置背压缓冲：最多缓存100条消息，溢出时打印日志并丢弃
                .onBackpressureBuffer(100, overflowMsg -> {
                    System.err.println("消息缓冲溢出，丢弃消息: " + new String(overflowMsg.getPayload()));
                });

        // 处理消息并批量确认
        messageFlux
                .doOnNext(message -> {
                    // 写入你的业务处理逻辑
                    String payload = new String(message.getPayload());
                    System.out.println("处理消息内容: " + payload);
                })
                // 每积累10条消息批量确认，减少API调用次数
                .buffer(10)
                .doOnNext(batch -> batch.forEach(AcknowledgeablePubsubMessage::ack))
                .subscribe();
    }
}

关键说明 #

• 背压机制：PubSubReactiveFactory返回的Flux遵循Reactive Streams规范，下游处理能力不足时会自动控制Pub/Sub消息拉取速率，避免大量消息堆积导致内存溢出。
• 背压策略：可通过onBackpressureBuffer、onBackpressureDrop、onBackpressureLatest等方法定义不同溢出处理逻辑，适配业务需求。
• 精细流量控制：若需更严格的速率限制，可使用limitRate(n)方法限制单次推送的消息数量：

  messageFlux
          .limitRate(50) // 每次最多推送50条消息
          .doOnNext(...)
  

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