Calling different bundles in parallel using ExecutorService

Question

I am working on a project in which I will be having different Bundles. Let's take an example, Suppose I have 5 Bundles and each of those bundles will have a method name process.

Below are the things, I am supposed to do-

1. I need to call all those 5 Bundles process method in parallel using multithreaded code and then write to the database. Meaning each bundle will return me back Map of String and String and then I am supposed to write that map into the database. I am not sure what is the right way to do that? Should I have five thread? One thread for each bundle? But what will happen in that scenario, suppose if I have 50 bundles, then I will have 50 threads?
2. And also, I want to have timeout feature as well. If any bundles is taking lot of time than the threshold setup by us, then it should get timeout and log as an error that this bundle has taken lot of time.

The following attempt that I have done is most probably flawed and error handling is by no means complete. And I am looking for best and efficient way of doing this problem.

Below is the solution I have. And below is my method which will call process method of all the bundles in a multithreaded way.

public void processingEvents(final Map<String, Object> eventData) throws InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(5);
    List<ProcessBundleHolderEntry> entries = new ArrayList<ProcessBundleHolderEntry>();

    Map<String, String> outputs = (Map<String, String>)eventData.get(BConstants.EVENT_HOLDER);

    for (BundleRegistration.BundlesHolderEntry entry : BundleRegistration.getInstance()) {
        ProcessBundleHolderEntry processBundleHolderEntry = new ProcessBundleHolderEntry(entry, outputs);
        entries.add(processBundleHolderEntry);
    }

    try {
        List<Future<Map<String, String>>> futures = pool.invokeAll(entries, 30L, TimeUnit.SECONDS);
        for (int i = 0; i < futures.size(); i++) {
            // This works since the list of future objects are in the
            // same sequential order as the list of entries
            Future<Map<String, String>> future = futures.get(i);
            ProcessBundleHolderEntry entry = entries.get(i);
            if (!future.isDone()) {
                // log error for this entry
            }
        }
    } catch (InterruptedException e) {
        pool.shutdownNow(); //cancels the tasks
        //restore interrupted flag and exit
        Thread.currentThread().interrupt();
        //or rethrow the exception
        throw e;
    }
}

Secondly, an implementation of Callable for your threads:

public class ProcessBundleHolderEntry implements Callable<Map<String, String>> {
    private BundleRegistration.BundlesHolderEntry entry;
    private Map<String, String> outputs;

    public ProcessBundleHolderEntry(BundleRegistration.BundlesHolderEntry entry, Map<String, String> outputs) {
        this.entry = entry;
        this.outputs = outputs;
    }

    public Map<String, String> call() throws Exception {
        final Map<String, String> response = entry.getPlugin().process(outputs);
        // write to the database.
        System.out.println(response);
        return response;
    }
}

Can anyone tell me whether there is any problem with the above approach or is there any better and efficient way of doing the same thing? I am not sure whether there is any thread safety issue as well.

Any help will be appreciated on this.

Answer 1

I need to call all those 5 Bundles process method in parallel using multithreaded code and then write to the database. I am not sure what is the right way to do that? Should I have five thread? One thread for each bundle? But what will happen in that scenario, suppose if I have 50 bundles, then I will have 50 threads?

It is recommended not to have more threads than the number of physical cpu's on that machine in order to be efficient. So if you have 50 Bundles and 8 cores for instances than an ExecutorService with 8 threads should be enough. The bundles will be executed 8 at a time and at most there will be 50/8 rounds of execution. (In real world some of the bundles will execute faster, etc.)

And also, I want to have timeout feature as well. If any bundles is taking lot of time than the threshold setup by us, then it should get timeout and log as an error that this bundle has taken lot of time.

You could use a Timer (ScheduledExecutor) on which you register the action to take place in case of a timeout. On timeout you should iterate the list of futures and cancel the ones that are not done yet.

Otherwise the code looks good enough. I can't see why should not be optimal.