Getting Feedback From Concurrent Tasks

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.