I have a method named process
in two of my Classes, lets say CLASS-A and CLASS-B
. Now in the below loop, I am calling process method
of both of my classes sequentially meaning one by one and it works fine but that is the not the way I am looking for.
for (ModuleRegistration.ModulesHolderEntry entry : ModuleRegistration.getInstance()) {
final Map<String, String> response = entry.getPlugin().process(outputs);
// write to database
System.out.println(response);
}
Is there any way, I can call the process method of both of my classes in a multithreaded way. Meaning one thread will call process method of CLASS-A and second thread will call process method of CLASS-B.
And then after that I was thinking to write the data that is being returned by the process
method into the database. So I can have one more thread for writing into database.
Below is the code that I came up with in a multithreaded way but somehow it is not running at all.
public void writeEvents(final Map<String, Object> data) {
// Three threads: one thread for the database writer, two threads for the plugin processors
final ExecutorService executor = Executors.newFixedThreadPool(3);
final BlockingQueue<Map<String, String>> queue = new LinkedBlockingQueue<Map<String, String>>();
@SuppressWarnings("unchecked")
final Map<String, String> outputs = (Map<String, String>)data.get(ModelConstants.EVENT_HOLDER);
for (final ModuleRegistration.ModulesHolderEntry entry : ModuleRegistration.getInstance()) {
executor.submit(new Runnable () {
public void run() {
final Map<String, String> response = entry.getPlugin().process(outputs);
// put the response map in the queue for the database to read
queue.offer(response);
}
});
}
Future<?> future = executor.submit(new Runnable () {
public void run() {
Map<String, String> map;
try {
while(true) {
// blocks until a map is available in the queue, or until interrupted
map = queue.take();
// write map to database
System.out.println(map);
}
} catch (InterruptedException ex) {
// IF we're catching InterruptedException then this means that future.cancel(true)
// was called, which means that the plugin processors are finished;
// process the rest of the queue and then exit
while((map = queue.poll()) != null) {
// write map to database
System.out.println(map);
}
}
}
});
// this interrupts the database thread, which sends it into its catch block
// where it processes the rest of the queue and exits
future.cancel(true); // interrupt database thread
// wait for the threads to finish
try {
executor.awaitTermination(5, TimeUnit.MINUTES);
} catch (InterruptedException e) {
//log error here
}
}
But If I remove the last line executor.awaitTermination(5, TimeUnit.MINUTES);
then it start running fine and after some time, I always get error like this-
JVMDUMP006I Processing dump event "systhrow", detail "java/lang/OutOfMemoryError" - please wait.
JVMDUMP032I JVM requested Heap dump using 'S:\GitViews\Stream\goldseye\heapdump.20130827.142415.16456.0001.phd' in response to an event
JVMDUMP010I Heap dump written to S:\GitViews\Stream\goldseye\heapdump.20130827.142415.16456.0001.phd
JVMDUMP006I Processing dump event "systhrow", detail "java/lang/OutOfMemoryError" - please wait.
Can anybody help me in figuring out what's the problem and what wrong I am doing in my above code? if I am running sequentially then I don't get any errors and it works fine.
And also is there any better way of doing this as compared to the way I am doing? Because in future I can have multiple plugin processor as compared to two.
What I am trying to do is- Call the process method of both of my classes in a multithreaded way and then write into the database bcoz my process method will return back a Map.
Any help will be appreciated on this.. And I am looking for a workable example on this if possible. Thanks for the help,