In my previous publication I showed example of C# parallel calculation where we assign task to the background thread and tell it which method to call when completed. There is one big advantage in C# over Java is that C# can pass pointers to the method as parameters, what allowed us to pass callback method which suppose to be invoked when result is ready. Unfortunately in Java we could not pass entire method as parameter to call it later, however there is workaround which I will show you today.
Task:
Execute parallel task which will invoke specified method on completion and return parameters from calculations to it.
A bit of additional information:
We will approach this task from the real live example. lets assume we are in the office where working several departments. We are Department Nr1 preparing tasks for other departments – “Management department” or “Caller”. Once we generate new task we passed it to the desired department – “Worker” and willing to be notified when “Worker” complete job so we could assign new job to it. It would mean that somebody from “Workers” department would have to come in in our office – “Call back method” and tell us that “Workers” ready to take new task and here is result from previous task. So lets look on implementation of such idea.
Implementation:
I will start from launcher:
package launcher;
import caller.Caller;
public class Launcher {
public static void main(String[] args) {
new Caller();
}
}
There is nothing to comment.
Next We will create interface:
package interfaces;
public interface CallBackInterface {
public void OnCallBackMethod(String returnedResult);
}
This interface will contain just one method – that is the one which will be our callBack method
Now lets look at the implementation of “Worker”
package workers;
import java.util.Random;
public class SlowWorker {
private int id;
private int timeToSleep;
public SlowWorker(int id)
{
this.id = id;
timeToSleep = new Random().nextInt(5000) + 1000;
}
public String doWork()
{
try{
Log("Worker ["+id+"] Starting Heavy Calculations");
Thread.sleep(timeToSleep);
int digit1 = new Random().nextInt(100);
int digit2 = new Random().nextInt(100);
int sum = digit1+digit2;
Log("Worker ["+id+"] Finished after: "+timeToSleep);
return "Worker: ["+id+"] Sum of "+digit1+" and "+digit2+" = "+sum;
}
catch(Exception e)
{
Log("I ["+id+"] was forcibly interrupted");
return "Worker: ["+id+"] Crash: "+e;
}
}
public void Log(String msg)
{
System.out.println(msg);
}
}
Worker accept own unique ID in constructor, just for understanding who is who, also it define sleep time as random from 1 sec to 6 sec, so each worker will be busy random amount of time
In doWork() method – we just sleep – pretending to do more heavy calculations and sum up two randomly generated digits. once it is done we return answer as String.
Now lets look at another worker – thats the one who will run from his “Workers” department to our “Management Department” and notify us that job was complete and give us text reports 🙂
package workers;
import interfaces.CallBackInterface;
import java.util.concurrent.Callable;
public class NotifierWorker implements Callable<Object>{
private CallBackInterface callBack;
private int id;
public NotifierWorker(int id)
{
this.id = id;
}
@Override
public Object call() throws Exception {
callBack.OnCallBackMethod(new SlowWorker(id).doWork());
return null;
}
public void setCallBack(CallBackInterface callBack)
{
this.callBack = callBack;
}
public CallBackInterface setCallBack()
{
return callBack;
}
}
This worker – “Notifier” implements Callable what automaticly generate call() method for us. that is where happens all the async calculations. however before we execute async calculation we have to set response method which is our Management office to specifically tell to Notifier worker where to find us – “Management” when his department would complete a job.
Lets have a look on that:
package caller;
import interfaces.CallBackInterface;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import workers.NotifierWorker;
public class Caller implements CallBackInterface{
private boolean workerComplete;
private int nrOfWorkersToRun;
private int jobsComplete;
private static int id;
public Caller()
{
workerComplete = false;
nrOfWorkersToRun = 5;
jobsComplete = 0;
id = 1;
startExecution();
}
@Override
public void OnCallBackMethod(String returnedResult) {
Log("Main Thread - OnCallBack: "+returnedResult);
jobsComplete++;
if(jobsComplete == 5)
workerComplete = true;
}
public void startExecution()
{
Log("Main Thread: Start Work @: " + new java.util.Date());
Log("");
ExecutorService es = Executors.newFixedThreadPool(3);
for(int i = 0; i < nrOfWorkersToRun; i++) {
int nr = getNextId();
Log("Main Thread: giving job to worker nr: ["+nr+"]");
NotifierWorker slaveNotifier = new NotifierWorker(nr);
slaveNotifier.setCallBack(this);
es.submit(slaveNotifier);
}
try
{
while(!workerComplete)
{
Log("Main Thread: doing my own stuff.........");
Thread.sleep(800);
}
}
catch(Exception e)
{
Log("Main Thread CRASHED: "+e);
}
Log("");
Log("Main Thread: End work @: " + new java.util.Date());
System.exit(0);
}
public int getNextId()
{
return id++;
}
public void Log(String msg)
{
System.out.println(msg);
}
}
As any management department it is big, important and hard to follow what a hell is it doing. But we will try:
in two words – we just schedule 5 tasks to workers and wait until all of them complete and displaying result of each completion once it is ready. Also keep in mind that we have 5 tasks however we limit our workers to only 3 concurrent task at a time to save up on their salaries 🙂
1) ExecutorService es = Executors.newFixedThreadPool(3); – We create Threadpool with size of three – that will allow us to limit all background tasks to three in total at the same time. (it is not a requirement, just more fun and extra chance to demonstrate thread pool queue)
2) NotifierWorker slaveNotifier = new NotifierWorker(nr); – We create notifiers and assign unique ID to each of them:
3) slaveNotifier.setCallBack(this); – Tell to each notifier where he can find us when job complete
4) es.submit(slaveNotifier); – send him to work on the given task
5) once hes task is complete he will tell us about it in OnCallBackMethod method where we keep track of how many was completed out of executed.
6) once amount of completed tasks equal to amount of scheduled tasks we assume that this is over and exit.
and here is console output for you:
Main Thread: Start Work @: Fri Mar 14 15:47:29 GMT 2014
Main Thread: giving job to worker nr: [1]
Main Thread: giving job to worker nr: [2]
Worker [1] Starting Heavy Calculations
Main Thread: giving job to worker nr: [3]
Worker [2] Starting Heavy Calculations
Main Thread: giving job to worker nr: [4]
Worker [3] Starting Heavy Calculations
Main Thread: giving job to worker nr: [5]
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Worker [2] Finished after: 2941
Main Thread - OnCallBack: Worker: [2] Sum of 19 and 58 = 77
Worker [4] Starting Heavy Calculations
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Worker [3] Finished after: 4552
Main Thread - OnCallBack: Worker: [3] Sum of 80 and 68 = 148
Worker [5] Starting Heavy Calculations
Worker [1] Finished after: 4794
Main Thread - OnCallBack: Worker: [1] Sum of 51 and 71 = 122
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Worker [4] Finished after: 5656
Main Thread - OnCallBack: Worker: [4] Sum of 81 and 46 = 127
Main Thread: doing my own stuff.........
Main Thread: doing my own stuff.........
Worker [5] Finished after: 5444
Main Thread - OnCallBack: Worker: [5] Sum of 1 and 93 = 94
Main Thread: End work @: Fri Mar 14 15:47:39 GMT 2014
And as usually you can download entire project here:
Vadims Briksins 