如何在Spring 5.1.3.RELEASE环境下无Java调度器读取GCP Pub/Sub消息

方案一：基于原生Google Cloud Pub/Sub客户端的长轮询拉取 #

直接使用google-cloud-pubsub原生客户端实现持续长轮询，替代Java调度器的定时拉取逻辑，客户端内部已封装长轮询机制，无需手动触发定时任务。

核心代码示例 #

import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import com.google.cloud.pubsub.v1.MessageReceiver;
import java.util.concurrent.Executors;

public class PubSubLongPollingConsumer {
    public static void initConsumer(String projectId, String subscriptionId) {
        ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId);

        // 消息处理逻辑
        MessageReceiver receiver = (PubsubMessage message, AckReplyConsumer consumer) -> {
            try {
                String messageContent = new String(message.getData().toByteArray());
                // 替换为你的业务处理代码
                System.out.println("Received message: " + messageContent);
                
                // 处理完成后手动确认消息
                consumer.ack();
            } catch (Exception e) {
                // 处理失败，触发消息重入队列
                consumer.nack();
                e.printStackTrace();
            }
        };

        // 创建订阅者，配置线程池处理消息
        Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
                .setExecutorProvider(Subscriber.defaultExecutorProviderBuilder()
                        .setExecutorThreadCount(4) // 根据业务压力调整线程数
                        .build())
                .build();

        // 启动订阅者，开始持续拉取
        subscriber.startAsync().awaitRunning();
        System.out.println("Pub/Sub subscriber started, listening for messages...");

        // 绑定Spring生命周期，关闭时停止订阅者
        Runtime.getRuntime().addShutdownHook(new Thread(subscriber::stopAsync));
    }
}

关键说明 #

• 原生Subscriber类内部实现了长轮询，无消息时会阻塞等待（可通过setPullTimeout配置超时时间），避免定时拉取的空轮询浪费资源
• 配合AckReplyConsumer手动控制消息确认/拒绝，保证消息处理的可靠性
• 在Spring 5.x环境中，可将该逻辑封装为Spring Bean，通过@PostConstruct启动、@PreDestroy停止，整合到容器生命周期

方案二：自定义Spring风格消息监听器容器 #

如果希望贴近Spring的消息监听范式，可自定义监听器容器封装原生客户端逻辑，适配Spring 5.1.x的生命周期管理。

核心步骤 #

1. 定义消息处理接口：

public interface PubSubMessageListener {
    void handleMessage(PubsubMessage message);
}

2. 实现监听器容器：

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;

public class PubSubListenerContainer implements InitializingBean, DisposableBean {
    private String projectId;
    private String subscriptionId;
    private PubSubMessageListener messageListener;
    private Subscriber subscriber;

    @Override
    public void afterPropertiesSet() throws Exception {
        ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId);
        subscriber = Subscriber.newBuilder(subscriptionName, (message, consumer) -> {
            try {
                messageListener.handleMessage(message);
                consumer.ack();
            } catch (Exception e) {
                consumer.nack();
                e.printStackTrace();
            }
        }).build();
        subscriber.startAsync().awaitRunning();
    }

    @Override
    public void destroy() throws Exception {
        if (subscriber != null) {
            subscriber.stopAsync().awaitTerminated();
        }
    }

    // getter/setter 省略
}

3. Spring配置注册Bean（以XML为例）：

<bean id="pubSubListenerContainer" class="com.yourpackage.PubSubListenerContainer">
    <property name="projectId" value="your-project-id"/>
    <property name="subscriptionId" value="your-subscription-id"/>
    <property name="messageListener" ref="customMessageListener"/>
</bean>

<bean id="customMessageListener" class="com.yourpackage.CustomMessageListenerImpl"/>

依赖配置 #

确保pom.xml中引入兼容Spring 5.1.x的Pub/Sub客户端版本：

<dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-pubsub</artifactId>
    <version>1.113.10</version> <!-- 该版本兼容Java 8及Spring 5.1.x，避免依赖冲突 -->
</dependency>

参考资料要点 #

• Google Cloud Pub/Sub Java客户端核心API：重点掌握Subscriber类的配置、消息确认机制、长轮询参数调整
• Spring 5.x Bean生命周期：熟悉InitializingBean和DisposableBean的用法，实现监听器的启停管理
• Pub/Sub可靠性最佳实践：包括消息重试策略、死信队列配置、批量拉取优化等

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