如何在Spring Boot应用中限制Google Pub/Sub消息消费速率？

方法一：拉取模式下主动控制消费速率 #

如果采用拉取模式（主动从Pub/Sub拉取消息），可以通过定时任务结合批量拉取的方式精准控制每分钟的消费总量：

1. 依赖spring-cloud-gcp-pubsub starter配置PubSubTemplate
2. 用Spring的@Scheduled定时任务，每分钟执行一次拉取逻辑，确保单次或多次拉取的总消息数不超过1000条

示例代码：

import com.google.pubsub.v1.PubsubMessage;
import org.springframework.cloud.gcp.pubsub.core.PubSubTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class RateControlledPullConsumer {

    private final PubSubTemplate pubSubTemplate;
    private static final String SUBSCRIPTION_NAME = "你的订阅名称";
    private static final int MAX_MESSAGES_PER_MINUTE = 1000;

    public RateControlledPullConsumer(PubSubTemplate pubSubTemplate) {
        this.pubSubTemplate = pubSubTemplate;
    }

    // 每分钟执行一次拉取
    @Scheduled(fixedRate = 60000)
    public void pullAndProcessMessages() {
        // 一次性拉取1000条，处理完成后手动ACK
        List<PubsubMessage> messages = pubSubTemplate.pull(SUBSCRIPTION_NAME, MAX_MESSAGES_PER_MINUTE, false);
        if (!messages.isEmpty()) {
            messages.forEach(this::processMessage);
            // 批量确认消息已处理
            pubSubTemplate.ack(SUBSCRIPTION_NAME, messages);
        }
    }

    private void processMessage(PubsubMessage message) {
        // 替换为你的业务处理逻辑
        System.out.println("处理消息：" + message.getData().toStringUtf8());
    }
}

注意：需要在启动类添加@EnableScheduling注解开启定时任务。

方法二：推送模式下限流处理 #

如果采用推送模式（Pub/Sub主动推送消息到应用），可以借助限流组件（比如Guava的RateLimiter）控制处理速率，确保每分钟最多处理1000条：

1. 引入Guava依赖（或其他限流组件）
2. 在消息消费方法中加入限流逻辑

示例代码：

import com.google.common.util.concurrent.RateLimiter;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.pubsub.v1.PubsubMessage;
import org.springframework.cloud.gcp.pubsub.core.PubSubTemplate;
import org.springframework.cloud.gcp.pubsub.integration.AckMode;
import org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

@Component
public class RateLimitedPushConsumer {

    // 设置速率：每分钟1000条 = 约16.67条/秒
    private final RateLimiter rateLimiter = RateLimiter.create(1000.0 / 60);

    @Bean
    public PubSubInboundChannelAdapter messageChannelAdapter(PubSubTemplate pubSubTemplate) {
        PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, "你的订阅名称");
        adapter.setOutputChannelName("pubsubInputChannel");
        adapter.setAckMode(AckMode.MANUAL); // 手动ACK，确保处理完成再确认
        return adapter;
    }

    @ServiceActivator(inputChannel = "pubsubInputChannel")
    public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
        // 等待获取处理许可，超过速率则阻塞
        rateLimiter.acquire();
        
        try {
            processMessage(message);
            consumer.ack(); // 处理成功后确认
        } catch (Exception e) {
            consumer.nack(); // 处理失败，让消息重新入队
        }
    }

    private void processMessage(PubsubMessage message) {
        // 替换为你的业务处理逻辑
        System.out.println("处理消息：" + message.getData().toStringUtf8());
    }
}

补充说明 #

• 拉取模式优势是精准控制拉取总量，避免消息堆积在应用端；
• 推送模式下的限流是在消费端控制处理速率，若Pub/Sub推送速率高于限流速率，消息会暂时堆积在订阅队列中（Pub/Sub会自动保存消息直到被确认）；
• 可根据实际场景调整拉取批次，比如将1000条分10次拉取（每次100条），避免单次拉取过多导致内存压力。

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