Multi threading with Java Executor

Question

Hi, I am stuck with this following problem. Say, I have a request which has 1000 items, and I would like to utilize Java Executor to resolve this.

Here is the main method

public static void main(String[] args) {

     //Assume that I have request object that contain arrayList of names
     //and VectorList is container for each request result

     ExecutorService threadExecutor = Executors.newFixedThreadPool(3);
     Vector<Result> vectorList = new Vector<Result();

     for (int i=0;i<request.size();i++) {
          threadExecutor.execute(new QueryTask(request.get(i).getNames, vectorList)
     }

      threadExecutor.shutdown();

      response.setResult(vectorList)

}

And here is the QueryTask class

public QueryTask() implements Runnable {

    private String names;
    private Vector<Result> vectorList;

    public QueryTask(String names, Vector<Result> vectorList) {
          this.names = names;
          this.vectorList = vectorList;
    }

    public void run() {
         // do something with names, for example, query database
         Result result = process names;

         //add result to vectorList
         vectorList.add(result);
    }


}

So, based on the example above, I want to make thread pool for each data I have in the request, run it simultaneously, and add result to VectorList. And at the end of the process, I want to have all the result already in the Vector list.

I keep getting inconsistent result in the response. For example, if I pass request with 10 names, I am getting back only 3 or 4, or sometimes nothing in the response. I was expecting if I pass 10, then I will get 10 back.

Does anyone know whats causing the problem?

Any help will be appreciate it.

Thanks

Answer 1

There are at least 2 issues here.

1. In your main, you shut down the ExecutorService, then try to get the results out right away. The executor service will execute your jobs asychronously, so there is a very good chance that all of your jobs are not done yet. When you call response.setResult(vectorList), vectorList is not fully populated.

2. You are concurrently accessing the same Vector object from within all of your runnables. This is likely to cause ConcurrentModificationExceptions, or just clobber stuff in the vector. You need to either manually synchronize on the vector inside of QueryTask, or pass in a thread-safe container instead, like Collections.synchronizedList( new ArrayList() );

Answer 2

You need to replace threadExecutor.shutdown(); with threadExecutor.awaitTermination();. After calling threadExecutor.shutdown(), you need to also call threadExecutor.awaitTermination(). The former is a nonblocking call that merely initiates a shutdown whereas the latter is a blocking call that actually waits for all tasks to finish. Since you are using the former, you are probably returning before all tasks have finished, which is why you don't always get back all of your results. The Java API isn't too clear, so someone filed a bug about this.

Answer 3

In addition to the answers already given, a little Agile refactoring will bring your code fully into Java 5 and make it more robust, reusable, and easier to use.

If you are going to use Executors, you should go all the way and use Futures as well. Change Runnable to Callable, and when you queue up your tasks, store the Future objects in your Vector (this should be a LinkedList, actually). The tasks should not have visibility into the storage mechanism for their results: this is too tightly coupled.

I also recommend not putting type names in your variable names (e.g. vectorList), and not using implementation names to store collections (e.g. Vector -- use the the List interface).

This code may be slightly more verbose, but is more loosely coupled and reusable. With a little more refactoring, it could be used for arbitrary tasks.

public QueryTask() implements Callable<Result> {

  private String names;

  public QueryTask(String names) {
    this.names = names;
  }

  public Result call() {
    // do something with names, for example, query database
    Result result = process names;

    //add result to vectorList
    vectorList.add(result);
  }
}

public static void main(String[] args) {
  ExecutorService exec = Executors.newFixedThreadPool(3);
  Collection<Future<Result>> results = new LinkedList<Result>();

  for (Object data : request) {
    results.add(threadExecutor.execute(new QueryTask(data)));
  }

  threadExecutor.awaitTermination();

  // Normally you would see if the Future is done, but
  // the previous statement did that implicitly.
  Collection<Result> finalResults = new LinkedList<Result>();
  for (Future<Result> result : results) {
    finalResults.add(result.get());
  }

  response.setResult(finalResults);
}

Answer 4

The easy solution is to add a call to ExecutorService.awaitTermination()

public static void main(String[] args) {

     //Assume that I have request object that contain arrayList of names
     //and VectorList is container for each request result

     ExecutorService threadExecutor = Executors.newFixedThreadPool(3);
     Vector<Result> vectorList = new Vector<Result();

     for (int i=0;i<request.size();i++) {
          threadExecutor.execute(new QueryTask(request.get(i).getNames, vectorList)
     }

      threadExecutor.shutdown();
      threadExecutor.awaitTermination(aReallyLongTime,TimeUnit.SECONDS);

      response.setResult(vectorList)

}