Showing posts with label CompletionService. Show all posts
Showing posts with label CompletionService. Show all posts

Thursday, December 30, 2010

Java5 CompletionService makes waiting for thread execution results so easy !

I recently had a scenario where I had to put some threads to do some work and wait till they complete. But I did not really want to wait till all of them completed their work, which is sub-optimal utilization of the cycles. I wanted to have some mechanism by which I could get the values as soon as each of my threads are done with the processing. So the scenario is something like this

  1. A main thread needs N tasks to be done 
  2. It spawns M threads from a Pool and works on the N tasks
  3. It waits till atleast one thread completes the task and does something with the result
  4. Continues to process the results till the time all of the N tasks are completed


    While I was breaking my head on this, I learnt about the "CompletionService" in Java5. This does exactly what I want, and I was able to complete the code in a few minutes. Attaching the code sample just in case this is useful to some.

    I defined a Callable which abstracts by business logic. You Can do what you need to do on the thread here

    package com.sreekanth.java5.threads;
    import java.util.concurrent.Callable;
    
    public class TestCallable implements Callable<Long> {
        @Override
        public Long call() throws Exception 
        {
            // do something here - just adding some numbers on the sample 
            
            long sum = 0;
            for (long i = 0; i <= 10; i++) 
            {
                sum += i;
            }
            return sum;
        }
    
    }
    
    

    Now I created the main class which does my processing using the CompletionService. The sample code is attached below

    package com.sreekanth.java5.threads;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class CompletionServiceTest 
    {
        /**
         * @param args
         */
        @SuppressWarnings("unchecked")
        
        public static void main(String[] args) 
        {
            CompletionService service = new ExecutorCompletionService(Executors.newFixedThreadPool(5));
            
            int noOfRunningFutures = 0;
            
            for (int i = 0; i < 20; i++) 
            {
                Callable<Long> worker = new TestCallable();
                service.submit(worker);
                
                // add them to the number of futures which I am creating - to keep track of the Queue length
                noOfRunningFutures ++;
            }
            
            while (noOfRunningFutures > 0) 
            {
                try {
                    
                    // this is a blocking call - whenever there is a worker which is already completed
                    // then it is fetched from the Queue                 
                    Future<Long> completed = service.take();
                    noOfRunningFutures --;
                    
                    // get the value from computed from the Future
                    Long i = completed.get();
                    
                    // Do some thing with the value got from the callable - just printing it here.
                    System.out.println("Got from the callable " + i);
                    
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }
    
    

    Really Cool and Simple!