Java concurrency – Feedback from 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.

Reference: Java concurrency examples – Getting feedback from concurrent tasks from our JCG partner Ricardo Zuasti at the Ricardo Zuasti’s blog blog.

Do you want to know how to develop your skillset to become a Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you two of our best selling eBooks for FREE!

JPA Mini Book

Learn how to leverage the power of JPA in order to create robust and flexible Java applications. With this Mini Book, you will get introduced to JPA and smoothly transition to more advanced concepts.

JVM Troubleshooting Guide

The Java virtual machine is really the foundation of any Java EE platform. Learn how to master it with this advanced guide!

Given email address is already subscribed, thank you!
Oops. Something went wrong. Please try again later.
Please provide a valid email address.
Thank you, your sign-up request was successful! Please check your e-mail inbox.
Please complete the CAPTCHA.
Please fill in the required fields.

Leave a Reply


eight × = 16



Java Code Geeks and all content copyright © 2010-2014, Exelixis Media Ltd | Terms of Use | Privacy Policy | Contact
All trademarks and registered trademarks appearing on Java Code Geeks are the property of their respective owners.
Java is a trademark or registered trademark of Oracle Corporation in the United States and other countries.
Java Code Geeks is not connected to Oracle Corporation and is not sponsored by Oracle Corporation.
Do you want to know how to develop your skillset and become a ...
Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you two of our best selling eBooks for FREE!

Get ready to Rock!
You can download the complementary eBooks using the links below:
Close