posted 29-Mar-2012 | 4 comments | , ,

Picking up from where I left off in my last post about the java.util.concurrent package, it’s interesting and sometimes mandatory to get feedback from concurrent tasks after they are started.

For example imagine an application that has to send email batches, besides from using a multi-threaded mechanism, you want to know how many of the intended emails were successfully dispatched, and during the actual sending process, the real-time progress of the whole batch.

To implement this kind of multi-threading with feedback we can use the Callable interface. This interface works mostly the same way as Runnable, but the execution method (call()) returns a value that should reflect the outcome of the performed computation.

Let’s first define the class that will perform the actual task:

package com.ricardozuasti;

import java.util.concurrent.Callable;

public class FictionalEmailSender implements Callable<Boolean> {
    public FictionalEmailSender (String to, String subject, String body){
        this.to = to;
        this.subject = subject;
        this.body = body;
    }

    @Override
    public Boolean call() throws InterruptedException {
        // Simulate that sending the email takes between 0 and 0.5 seconds
        Thread.sleep(Math.round(Math.random()* 0.5 * 1000));

        // Lets say we have an 80% chance of successfully sending our email
        if (Math.random()>0.2){
            return true;
        } else {
            return false;
        }
    }

    private String to;
    private String subject;
    private String body;
}

Notice that your Callable can use any return type, so your task can return whatever info you need.

Now we can use a thread pool ExecutorService to send our emails, and since our task is implemented as a Callable, we get a Future reference for each new task we submit for execution. Note that we will create our ExecutorService using a direct constructor instead of a utility method from Executors, this is because using the specific class (ThreadPoolExecutor) provides some methods that will come in handy (not present present in the ExecutorService interface).

package com.ricardozuasti;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Concurrency2 {

    public static void main(String[] args) {
        try {
            ThreadPoolExecutor executor = new ThreadPoolExecutor(30, 30, 1, TimeUnit.SECONDS,
                    new LinkedBlockingQueue());

            List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(9000);

            // Lets spam every 4 digit numeric user on that silly domain
            for (int i = 1000; i < 10000; i++) {
                futures.add(executor.submit(new FictionalEmailSender(i + "@wesellnumericusers.com",
                        "Knock, knock, Neo", "The Matrix has you...")));
            }

            // All tasks have been submitted, wen can begin the shutdown of our executor
            System.out.println("Starting shutdown...");
            executor.shutdown();

            // Every second we print our progress
            while (!executor.isTerminated()) {
                executor.awaitTermination(1, TimeUnit.SECONDS);
                int progress = Math.round((executor.getCompletedTaskCount() * 100) /
                                          executor.getTaskCount());

                System.out.println(progress + "% done (" + executor.getCompletedTaskCount() +
                                   " emails have been sent).");
            }

            // Now that we are finished sending all the emails, we can review the futures
            // and see how many were successfully sent
            int errorCount = 0;
            int successCount = 0;
            for (Future<Boolean> future : futures) {
                if (future.get()) {
                    successCount++;
                } else {
                    errorCount++;
                }
            }

            System.out.println(successCount + " emails were successfully sent, but "
                    + errorCount + " failed.");

        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

After all tasks are submitted to the ExecutorService, we begin it’s shutdown (preventing new tasks from being submitted) and use a loop (in a real-life scenario you should continue doing something else if possible) to wait until all tasks are finished, calculating and printing the progress made so far on each iteration. Note that you could store the executor reference and query it from other threads any time to calculate and report the process progress.

Finally, using the collection of Future references we got for each Callable submitted to the ExecutorService, we can inform the number of emails successfully sent and the number that failed to.

This infrastructure is not only easy to use but also promotes clear separation of concerns, providing a pre-defined communication mechanism between the dispatcher program and the actual tasks.

  • Anonymus

    Nice Article.. Can you explain when do we use ExecutorCompletionService

  • http://ricardozuasti.com/ Ricardo Zuasti

    The ExecutorCompletionService can be used when you have a collection of Callables to run and want to get the results to check/use after they are completed. They key to it is the take() method, which blocks your thread until a result is available to be returned. 

    Check out the samples in the Javadoc (http://docs.oracle.com/javase/1.5.0/docs/api/java/util/concurrent/ExecutorCompletionService.html), they are quite clear imho.

  • Glyn Normington

    Just wanted to bring up an issue which relates to this topic. The behaviour of ThreadPoolExecutor.getCompletedTaskCount seems to have changed in Java 7. Previously it excluded tasks that failed with an exception, but now it includes such tasks. Do you think that’s a bug? Or was it really a bug in Java 6 and earlier?

  • http://ricardozuasti.com/ Ricardo Zuasti

    The API documentation didn’t change at all from Java 6 to 7, and there is no way to get the number of failed tasks in either version, so it seems completely feasible that they changed the behavior of the method. I wouldn’t qualify either as a bug since “complete” is ambiguous enough to include only successfully finished tasks or all finished tasks (failed and successful).

    I guess counting failures should be done manually if you want a precise report on the status of all completed tasks.