I have an async chain in my java code that i want to stop after a certain timeout
so i created a threadPool with some threads and called the CompletableFuture like this
ExecutorService pool = Executors.newFixedThreadPool(10);
than i have a cyclic method that loads data from the db and executes some task on it, once all the CompletableFutures are completed its doing it again
CompletableFuture<MyObject> futureTask =
CompletableFuture.supplyAsync(() -> candidate, pool)
.thenApply(Task1::doWork).thenApply(Task2::doWork).thenApply(Task3::doWork)
.thenApply(Task4::doWork).thenApply(Task5::doWork).orTimeout(30,TimeUnit.SECONDS)
.thenApply(Task6::doWork).orTimeout(30,TimeUnit.SECONDS)
.exceptionally(ExceptionHandlerService::handle);
My problem is in task6, that has a very intensive task (its a network connection task that sometimes hangs forever)
i noticed that my orTimeout is being fired correctly after 30 seconds, but the thread running Task6 is still being running
after few cycles like this, all my threads are drained and my app dies
How can i cancel the running threads on the pool after the timeout has reached?
(without calling pool.shutdown())
UPDATE*
inside the main thread i did a simple check as shown here
for (int i = TIME_OUT_SECONDS; i >= 0; i--) {
unfinishedTasks = handleFutureTasks(unfinishedTasks, totalBatchSize);
if(unfinishedTasks.isEmpty()) {
break;
}
if(i==0) {
//handle cancelation of the tasks
for(CompletableFuture<ComplianceCandidate> task: unfinishedTasks) {
**task.cancel(true);**
log.error("Reached timeout on task, is canceled: {}", task.isCancelled());
}
break;
}
try {
TimeUnit.SECONDS.sleep(1);
} catch (Exception ex) {
}
}
What i see is that after few cycles, all the tasks complain about timeout...
in the first 1-2 cycles, i still get epected responses (while there are threads to process it)
i still feel that the thread pool is exhausted
See Question&Answers more detail:
os 与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…