如何检查 gcp pubsub 空/非活动订阅答案

【问题标题】：How to check gcp pubsub empty/inactive subscription如何检查 gcp pubsub 空/非活动订阅
【发布时间】：2021-07-12 18:51:19
【问题描述】：

我有一个订阅 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());
    }
}

【问题讨论】：

你为什么要停止你的池？你有什么问题？
释放 pod 上的资源。当我不这样做时，JVM 永远不会释放线程池资源。

标签： java spring multithreading google-cloud-platform review

【解决方案1】：

您的池占用很少的空间和内存，并且在不使用时几乎不消耗 CPU。为您的池容量设置一个最大限制，并在尝试缩减它时使用它。如果您有太多消息要处理，任务会排队等待空闲的执行器池来完成任务。

如果您有上下可扩展性方面的问题，您的设计可能会被审核。您可以在集群中触发事件并在其他 pod 上并行处理它们，而不是 pod 内部的 executorPool。这些 pod 将能够根据流量扩大和缩小（查看 Knative）

【讨论】：