How to use ExecutorService to poll until a result

问题:

I have a scenario where I have to poll a remote server checking if a task has completed. Once it has, I make a different call to retrieve the result.

I originally figured I should use a SingleThreadScheduledExecutor with scheduleWithFixedDelay for polling:

ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture future = executor.scheduleWithFixedDelay(() -> poll(jobId), 0, 10, TimeUnit.SECONDS);

public void poll(String jobId) {
   boolean jobDone = remoteServer.isJobDone(jobId);
   if (jobDone) {
       retrieveJobResult(jobId);
   }
}

But since I can only provide a Runnable to scheduleWithFixedDelay which can't return anything, I don't understand when the future will be complete, if ever. What does calling future.get() even mean? What result am I waiting for?

The first time I detect the remote task has completed, I want to execute a different remote call and set its result as the value of the future. I figured I could use CompletableFuture for this, that I would forward to my poll method, which would in turn forward it to my retrieveTask method that would eventually complete it:

CompletableFuture<Object> result = new CompletableFuture<Object>();
ScheduledFuture future = executor.scheduleWithFixedDelay(() -> poll(jobId, result), 0, 10, TimeUnit.SECONDS);

public void poll(String jobId, CompletableFuture<Object> result) {
   boolean jobDone = remoteServer.isJobDone(jobId);
   if (jobDone) {
       retrieveJobResult(jobId, result);
   }
}

public void retrieveJobResult(String jobId, CompletableFuture<Object> result) {
    Object remoteResult = remoteServer.getJobResult(jobId);
    result.complete(remoteResult);
}

But this has a ton of issues. For one, CompletableFuture doesn't even seem to be intended for this kind of use. Instead I should be doing CompletableFuture.supplyAsync(() -> poll(jobId)) I think, but how would I then properly shutdown the executor and cancel the future it returned when my CompletableFuture is canceled/complete? It feels like polling should be implemented in some completely different way.

回答1:

I think CompletableFutures are a fine way to do this:

ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

private void run() {
    final Object jobResult = pollForCompletion("jobId1")
            .thenApply(jobId -> remoteServer.getJobResult(jobId))
            .get();

}

private CompletableFuture<String> pollForCompletion(String jobId) {
    CompletableFuture<String> completionFuture = new CompletableFuture<>();
    final ScheduledFuture<Void> checkFuture = executor.scheduleAtFixedRate(() -> {
        if (remoteServer.isJobDone(jobId)) {
            completionFuture.complete(jobId);
        }
    }, 0, 10, TimeUnit.SECONDS);
    completionFuture.whenComplete((result, thrown) -> {
        checkFuture.cancel(true);
    });
    return completionFuture;
}

回答2:

it seems to me you are more worried by some stylistic problems than any others. in java 8, CompletableFuture has 2 roles: one is the traditional future, which gives an asynchronous source for task execution and status query; the other is what we usually call a promise. a promise, if you don't know yet, can be considered a builder of future and its completion source. so in this case, intuitively a promise is required, which is the exact case you are using here. the examples you are worrying about is something that introduces you the first usage, but not the promise way.

accepting this, it should be easier for you to start dealing with your actual problem. i think the promise is supposed to have 2 roles, one is to notify your task completion of polling, and the other is to cancel your scheduled task on completion. here should be the final solution:

public CompletableFuture<Object> pollTask(int jobId) {
    CompletableFuture<Object> fut = new CompletableFuture<>();
    ScheduledFuture<?> sfuture = executor.scheduleWithFixedDelay(() -> _poll(jobId, fut), 0, 10, TimeUnit.SECONDS);
    fut.thenAccept(ignore -> sfuture.cancel(false));
    return fut;
}

private void _poll(int jobId, CompletableFuture<Object> fut) {
    // whatever polls
    if (isDone) {
        fut.complete(yourResult);
    }
}