java - 如何检查 gcp pubsub 空/非活动订阅

问题描述

我有一个订阅 GCP 中的主题的应用程序，当那里有一些消息时，它会下载它们并将它们发送到 ActiveMQ 上的队列。

为了使这个过程更快，我正在使用 executorService 并启动多个线程来向 activeMQ 发送消息。由于这个订阅应该是一个持续的任务，我将代码放在一个 while(true) 循环中，因此我无法以正常方式关闭 executorService，因为我将创建和关闭 executor 服务每个循环。

我正在寻找一种优雅的方式来在订阅为空（主题中没有数据）2 或 3 分钟或某个不活动窗口时关闭 executorService。然后当有一些新数据时它当然会重新开始。

以下是我不喜欢的想法，这只是我在订阅未检索到数据时递增的计数器。

我正在寻找一种更优雅的方式来做到这一点。

@Service
@Slf4j
public class PubSubSubscriberService {

private static final int EMPTY_SUBSCRIPTION_COUNTER = 4;
private static final Logger businessLogger = LoggerFactory.getLogger("BusinessLogger");
private Queue<PubsubMessage> messages = new ConcurrentLinkedQueue<>();



public void pullMessagesAndSendToBroker(CompositeConfigurationElement cce) {

    var patchSize = cce.getSubscriber().getPatchSize();
    var nThreads = cce.getSubscriber().getSendingParallelThreads();
    var scheduledTasks = 0;
    var subscribeCounter = 0;
    ThreadPoolExecutor threadPoolExecutor = null;


    while (true) {
       try {
           if (subscribeCounter < EMPTY_SUBSCRIPTION_COUNTER) {
                log.info("Creating Executor Service for uploading to broker with a thread pool of Size: " + nThreads);
            threadPoolExecutor = getThreadPoolExecutor(nThreads);
        }

        var subscriber = this.getSubscriber(cce);
        this.startSubscriber(subscriber, cce);
        this.checkActivity(threadPoolExecutor, subscribeCounter++);


        // send patches of {{ messagesPerIteration }}
        while (this.messages.size() > patchSize) {
            if (poolIsReady(threadPoolExecutor, nThreads)) {
                UploadTask task = new UploadTask(this.messages, cce, cf, patchSize);
                threadPoolExecutor.submit(task);
                scheduledTasks ++;
            }
            subscribeCounter = 0;
        }

        // send the rest
        if (this.messages.size() > 0) {
            UploadTask task = new UploadTask(this.messages, cce, cf, patchSize);
            threadPoolExecutor.submit(task);
            scheduledTasks ++;
            subscribeCounter = 0;
        }

        if (scheduledTasks > 0) {
            businessLogger.info("Scheduled " + scheduledTasks + " upload tasks of size upto: " + patchSize + ", preparing to start subscribing for 30 more sec") ;
            scheduledTasks = 0;
        }
    } catch ( Exception e) {
        e.printStackTrace();
        businessLogger.error(e.getMessage());
    }
}

标签： javaspringmultithreadinggoogle-cloud-platformreview

java - 如何检查 gcp pubsub 空/非活动订阅

问题描述

解决方案

推荐阅读