Example : Using Java future, Callable, Executor and CountDownLatch

When you start using java.util.concurrent’s Future and the  thread pools or executor framework, Following questions may come to one’s mind

• How to do thread join kind of operation ?
• How to set timeout for future?
• Can the futures notify back the submitting thread when they are done?

Example explaining usage for future, Callable and CountDownLatch together

 
public class DataAccessor {
    private static ThreadPoolExecutor executor;
    private int timeout = 1000;
    static {
        executor = new ThreadPoolExecutor(10, 10, 1000,     TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1000));
     }
    public ProcessedResponse getDataFromService(List<String> requests) {
        final CountDownLatch latch = new CountDownLatch(requests.size());
        List<Future<SubmittedJob>> submittedJobs = new ArrayList<Future<SubmittedJob>>(
            requests.size());
        for (String request : requests) {
            Future<SubmittedJob> job = executor.submit(new GetAndProcessResponse(request, latch));
            submittedJobs.add(job);
       }
       if (!latch.await(timeout, TimeUnit.MILLISECONDS)) {
            // some of the jobs not done
       }
       for (Future<SubmittedJob> job : submittedJobs) {
           try {
               // before doing a get you may check if it is done
               if (!job.isDone()) {
                   // cancel job
                   job.cancel(true);
                   continue;
              }
              ProcessedResponse response = job.get();
              aggregateResponse(response);
         } catch (ExecutionException cause) {
           // exceptions occurred during execution
           // we can get cause.getCause() and check instanceof
        }
    }
 }
private void aggregateResponse(ProcessedResponse response) {
    // aggregate response to a single response.
 }
private class SubmittedJob {
    final Future<ProcessedResponse> job;
    public Future<ProcessedResponse> getJob() {
        return job;
    }
    public String getRequest() {
        return request;
    }
     final String request;
     SubmittedJob(final Future<ProcessedResponse> job, final String request) {
         this.job = job;
         this.request = request;
     }
 }
private class ProcessedResponse {
    ProcessedResponse(final String request) {
    }
 }
private class GetAndProcessResponse implements Callable<ProcessedResponse> {
    private final String request;
    private final CountDownLatch countDownLatch;
    GetAndProcessResponse(final String request,    final CountDownLatch countDownLatch) {
        this.request = request;
        this.countDownLatch = countDownLatch;
    }
    ProcessedResponse call() {
         try {
             return getAndProcessResponse(this.request);
         } finally {
             countDownLatch.countDown();
         }
     }
    private ProcessedResponse getAndProcessResponse(final String request) {
         // do the service call
         // ........
         return (new ProcessedResponse(request));
    }
}

Advertisement