【发布时间】:2020-09-11 18:31:44
【问题描述】:
问题
我正在使用完成服务并生成子线程来执行一些 ETL。 当我在 IDE 中调试然后停止所有进程时,我注意到我仍然有一堆僵尸线程正在杀死我的 CPU。这是因为我没有正确终止子线程。
最小示例
Future<Boolean> future = completionService.submit(conversionProcessor);
boolean isCompleted = false;
while (!isCompleted && !closed.get()) {
try {
isCompleted = future.get(CONSUMER_HEARTBEAT_INTERVAL,
TimeUnit.SECONDS); // Wait until heartbeat interval exceeds
if (isCompleted) {
// do some things here
future.cancel(true);
break;
}
} catch (TimeoutException e) {
// Used to keep consumer alive in the cluster
consumer.poll(Duration.ofSeconds(CONSUMER_HEARTBEAT_INTERVAL)); // does heart-beat
} catch (CancellationException e) {
future.cancel(true);
break;
} catch (InterruptedException e) {
future.cancel(true);
break;
} catch (WakeupException we) {
future.cancel(true);
break;
} catch (Exception e) {
future.cancel(true);
break;
}
想法
基本上,我将我的Callable<Boolean> 提交给我的完成服务。
ExecutorCompletionService<Boolean> completionService = new ExecutorCompletionService<>(
Executors.newSingleThreadExecutor());
如果我停止调试器,该线程可能仍在运行。我刚刚添加了这个 future.cancel(true) 片段,它似乎已经停止从我的子线程连续上传文件,但我仍然看到这些 java 进程在我的活动监视器上运行。
我想知道我应该如何考虑这个问题?我想要可调用对象,因为它告诉我 何时 底层 ETL 是否已完成(真/假)
编辑: future.cancel 实际上似乎有很大帮助.. 这是我想要使用的吗?
【问题讨论】:
-
请注意,在一个单独的线程中运行您的conversionProcessor,然后在同时保持主线程忙碌以及在while 循环中检查其他线程是否已完成其工作是非常低效的。您现在保持两个线程被占用,它可能是一个。更好的是,例如,只使用 CompletableFuture,然后根据 CompletableFuture.thenAccept 中的过程结果组合您想做的任何工作。
-
我必须让主线程在一个while循环中,因为Kafka集群需要心跳,如果超过3秒,长时间运行的ETL会导致消费者组重新平衡。
标签: java multithreading executorservice completion-service