Part 7: java.util.concurrent - invokeAll via ExecutorService

Submit Collection of Threads via ExecutorService

Today in next part of the series we will talk about How we can release multiple threads via Executor Interface.

As per JAVA 6.0 docs –

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)                                                            throws InterruptedException

In traditional Java – If we have to release multiple threads- we have to create Thread objects and call

Start method one by one.

In Java 5.0 and above – If we can put all callable objects in a collection object and pass the collection objects to ExecutorService to release.

For example – If we have 1000 thread object to be released, we can create an array list and an ExecutorService object with thread pool size of 20.

        // create an ExecutorService object with thread pool size = 20

        // create an array List with 1000 callable objects.

        // Release all threads by ExecutorService.

Some important points to note here are

• Ø  As ExecutorService has thread pool of 20, it will internally manage queuing and releasing of 1000 threads.
• Ø  invokeAll is a blocking method. It means – JVM won’t proceed to next line until all the 1000 threads are complete.
• Ø  All the 1000 threads must have same Return Type < T >.
• Ø  Future.isDone() is true for each element of the returned list
• Ø  A completed thread could have terminated either normally or by throwing an exception. In both the case return type of Future.isDone( ) is true.
• Ø  The results of this method are undefined if the given collection is modified while this operation is in progress
• Ø  It can throw following exceptions –
• InterruptedException - if interrupted while waiting, in which case unfinished tasks are cancelled.
• NullPointerException - if tasks or any of its elements are null.
• RejectedExecutionException - if any task cannot be scheduled for execution.

EXAMPLE –

Imagine a scenario, one needs to write create a txn file of each request coming to system. Following are the requirements-

• Ø  File name will be the client name.
• Ø  At a time, almost 1000 clients could be connected at a time, so system must be capable of creating 1000 files in one go.
• Ø  Process of creating one file should be less than 5 ms.
• Ø  Total time of creating all the files should be less than 300 ms.

Let’s write the pseudo code for this case –

File creation Thread – implements CALLABLE<Boolean>

// GET the File name and Raw Data in Constructor

Try{

// Create a File Object

// Create a print Writer Object

// Prepare the data and Write in the File

// Close the print Writer and File Object

// Return TRUE

} catch Exception {

return FALSE

}

                            

Main Parent Thread

  // -- In the Loop for whole Client list ---
    try{

        // create a file creator thread – Pass filename and raw Data

// add the thread object into an Array List

// Pass the array list to Executor Service interface and invokeAll.

// Loop in the Future<Boolean> and check how many threads completed successfully.

  }catch Exception{

// Take necessary action

}finally{

// Shut down the ExecutorService
 }

package com.jovialjava.blog.threads;

import java.io.File;
import java.io.PrintWriter;
import java.util.*;
import java.util.concurrent.*;


// FILE CREATOR THREAD
class FileTask implements Callable< Boolean >{
 
 private String name = null;
 private String data = null;
 public FileTask(String name, String data){
  this.name = name;
  this.data = data;
 }
 
 public Boolean call(){
  try{
   File file = new File(this.name);
   PrintWriter writer = new PrintWriter(file);
   writer.write(this.data);
   writer.close();
   return true;   
  }catch(Exception e){
   return false;
  }
 }
}

// MAIN THREAD
public class InvokeAllExample {
 
 private static final ExecutorService executorPool=Executors.newFixedThreadPool(20);
 private static final int NO_OF_CLIENT = 1000;
 private static final String FILE_EXT = ".txt";
 private static String TXN_DATA = "SOME RANDOM TXN DATA FOR CLIENT --> ";
 private static String DIRECTORY = "EXAMPLE" + File.separator;
 
 static{
  if(!new File(DIRECTORY).isDirectory()){
   new File(DIRECTORY).mkdir();
  }
 }
 
 public static void main(String[] args) {
  int success = 0;
  int failure = 0;
  /**
   * Lets assume we have 1000 clients connected and sending request at a time.
   */
  Collection collection = new ArrayList( );
  for(int i=0; i< NO_OF_CLIENT; i++){
   FileTask task = new FileTask(DIRECTORY+ Integer.toString(i)+ FILE_EXT, TXN_DATA+ i);
   collection.add(task);
  }
  long startTime = new Date().getTime();
  try {   
   List< Future< Boolean > > list = executorPool.invokeAll(collection);   
    for(Future< Boolean > fut : list){
     int ignore = fut.get()? success++ : failure++;
    }
   
   } catch (Exception e) {
     e.printStackTrace();
   }finally{
     System.out.println("TOTAL SUCCESS - "+ success);
     System.out.println("TOTAL FAILURE - "+ failure);
     System.out.println("Total time - " + (new Date().getTime() - startTime) + " ms");
     executorPool.shutdown();
   }
 } // End of Main

}