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-
- 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? - 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.