Blocking queues in java

The key facilities that BlockingQueue provides to such systems are, as its name implies, enqueuing and dequeueing methods that do not return until they have executed successfully. So, for example, a print server does not need to constantly poll the queue to discover whether any print jobs are waiting; it need only call the poll method, supplying a timeout, and the system will suspend it until either a queue element becomes available or the timeout expires. BlockingQueue defines seven new methods, in three groups:

Group1 : Adding an Element

boolean offer(E e, long timeout, TimeUnit unit)
                // insert e, waiting up to the timeout
void put(E e)   // add e, waiting as long as necessary

The nonblocking overload of offer defined in Queue will return false if it cannot immediately insert the element. This new overload waits for a time specified using java.util.concurrent.TimeUnit, an Enum which allows timeouts to be defined in units such as milliseconds or seconds.

Taking these methods together with those inherited from Queue, there are four ways in which the methods for adding elements to a BlockingQueue can behave: offer returns false if it does not succeed immediately, blocking offer returns false if it does not succeed within its timeout, add throws an exception if it does not succeed immediately, and put blocks until it succeeds.

Group2  : Removing an Element

E poll(long timeout, TimeUnit unit)
                // retrieve and remove the head, waiting up to the timeout
E take()        // retrieve and remove the head of this queue, waiting
                // as long as necessary

Again taking these methods together with those inherited from Queue, there are four ways in which the methods for removing elements from a BlockingQueue can behave: poll returns null if it does not succeed immediately, blocking poll returns null if it does not succeed within its timeout, remove throws an exception if it does not succeed immediately, and take blocks until it succeeds.

Group 3 : Retrieving or Querying the Contents of the Queue

int drainTo(Collection<? super E> c)
                // clear the queue into c
int drainTo(Collection<? super E> c, int maxElements)
                // clear at most the specified number of elements into c
int remainingCapacity()
                // return the number of elements that would be accepted
                // without blocking, or Integer.MAX_VALUE if unbounded

The drainTo  methods perform atomically and efficiently, so the second overload is useful in situations in which you know that you have processing capability available immediately for a certain number of elements, and the first is useful for example when all producer threads have stopped working. Their return value is the number of elements transferred. RemainingCapacity reports the spare capacity of the queue, although as with any such value in multi-threaded contexts, the result of a call should not be used as part of a test-then-act sequence; between the test (the call of remainingCapacity) and the action (adding an element to the queue) of one thread, another thread might have intervened to add or remove elements.

BlockingQueue guarantees that the queue operations of its implementations will be thread-safe and atomic.
But this guarantee doesn't extend to the bulk operations inherited from CollectionaddAll, containsAll, retainAll and removeAllunless the individual implementation provides it. So it is possible, for example, for addAll to fail, throwing an exception, after adding only some of the elements in a collection.

Blocking queue has the following characteristics:

• methods to add an item to the queue, waiting for space to become available in the queue if necessary;
  
• corresponding methods that take an item from the queue, waiting for an item to put in the queue if it is empty;
  
• optional time limits and interruptibility on the latter calls;
  
• efficient thread-safety: blocking queues are specifically designed to have their put() method called from one thread and the take() method from another— in particular, items posted to the queue will be published correctly to any other thread taking the item from the queue again; significantly, the implementations generally achieve this without locking the entire queue, making them highly concurrent components;
  
• integration with Java thread pools: a flavour of blocking queue can be passed into the constructor of ThreadPoolExecutor to customise the behaviour of the thread pool.

Implementations of blocking queue
ArrayBlockingQueue : A simple bounded BloickingQueue implementation backed by an array.

DelayQueue : An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired.It uses elements that implement the new java.util.concurrent.Delayed interface.

PriorityBlockingQueue : This queue bases ordering on a specified Comparator, and the element returned by any take( ) call is the smallest element based on this ordering.

LinkedBlockingQueue : A simple bounded BloickingQueue implementation backed by a linked list.

SynchronousQueue : This queue has a size of zero (yes, you read that correctly). It blocks put( ) calls until another thread calls take( ), and blocks take( ) calls until another thread calls put( ). Essentially, elements can only go directly from a producer to a consumer, and nothing is stored in the queue itself (other than for transition purposes).


Example - Producer consumer problem with Blocking queue
The queue takes care of all the details of synchronizing access to its contents and notifying other threads of the availability of data.

Producer.java

public class Producer extends Thread {
    private BlockingQueue cubbyhole;
    private int number;
                          
    public Producer(BlockingQueue c, int num) {
        cubbyhole = c;
        number = num;
    }

    public void run() {
        for (int i = 0; i < 10; i++) {
            try {
                cubbyhole.put(i);
                System.out.format("Producer #%d put: %d%n", number, i);
                sleep((int)(Math.random() * 100));
            } catch (InterruptedException e) { }
        }
    }
}

Consumer.java

import java.util.concurrent.*;
public class Consumer extends Thread {
    private BlockingQueue<Integer> cubbyhole;
    private int number;
        
    public Consumer(BlockingQueue<Integer> c, int num) {
        cubbyhole = c;
        number = num;
    }

    public void run() {
        int value = 0;
        for (int i = 0; i < 10; i++) {
            try {
                value = cubbyhole.take();
                System.out.format("Consumer #%d got: %d%n", number, value);  
            } catch (InterruptedException e) { }        
        }
    }
}

ProducerConsumerTest.java

public class ProducerConsumerTest {
    public static void main(String[] args) {

        ArrayBlockingQueue c = new ArrayBlockingQueue(1);
        Producer p1 = new Producer(c, 1);
        Consumer c1 = new Consumer(c, 1);

        p1.start();
        c1.start();
    }
}

Possible Use cases for BlockingQueue

These features make BlockingQueues useful for cases such as the following:

• a server, where incoming connections are placed on a queue, and a pool of threads picks them up as those threads become free;
  
• in a variety of parallel processes, where we want to manage or limit resource usage at different stages of the process.

invokeAll via ExecutorService

Syntax of this method is like this(in java 6):

<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.

The invokeAll() method invokes all of the Callable objects you pass to it in the collection passed as parameter. The invokeAll() returns a list of Future objects via which you can obtain the results of the executions of each Callable. invokeAll is a blocking method. It means – JVM won’t proceed to next line until all the threads are complete.


Keep in mind that a task might finish due to an exception, so it may not have "succeeded". There is no way on a Future to tell the difference.

Example:

ExecutorService executorService = Executors.newFixedThreadPool();

List<Callable<String>> callables = new ArrayList<Callable<String>>();

callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});

List<Future<String>> futures = executorService.invokeAll(callables);

for(Future<String> future : futures){
    System.out.println("future.get = " + future.get());
}

executorService.shutdown();

ArrayBlockingQueue

An ArrayBlockingQueue is a queue returned by an array that have a limitation. It is a "bounded buffer" in which elements are helds in a constant size array. Once the capacity of this queue is defined it can not be grew up, if you will try to put element into the full queue will result in a blocking wait and similarly, obtain the element from vacate queue will block. In this queue the elements are ordered in FIFO (First-In-First-Out). In this queue the element that has been the highest time on the queue is the head element, and the element that has been the minimum time on the queue is the tail element of the queue. Insertion of an element in this queue is happened at tail and the element is retrieved from the head position.
All of the optional methods of the Collection and Iterator interfaces are implemented by this class and its iterator.

Syntax

public class ArrayBlockingQueue<E>
Parameter description
E : It is the element's type that's held in this collection.
Constructor of ArrayBlockingQueue class are :

• ArrayBlockingQueue(int capacity) : With the constant capability and default way of accessing this constructor makes an ArrayBlockingQueue.
• ArrayBlockingQueue(int capacity, boolean fair) : With the constant capability and the specified way of accessing this constructor makes an ArrayBlockingQueue.
• ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) : With the constant capability, the specified way of accessing and the elements of the given collection this constructor makes an ArrayBlockingQueue, that are appended in collection's iterator traversing order.

Example :
Here we give a simple example which will illustrate you to how can you use the methods of ArrayBlockingQueue class.

public class ArrayBlockingQueueDemo {
  public static void main(String args[]) {
    ArrayBlockingQueue abq = new ArrayBlockingQueue(10);
    abq.add(1);
    abq.add(2);
    abq.add(3);
    abq.add(4);
    abq.add(5);
    System.out.println("Elements of queue1= " + abq);
    ArrayBlockingQueue abq1 = new ArrayBlockingQueue(10);
    abq1.offer("A");
    abq1.offer("B");
    abq1.offer("C");
    abq1.offer("D");
    abq1.offer("E");
    abq1.offer("F");
    System.out.println("Elements of queue2 = " + abq1);
    int i = abq.drainTo(abq1, 4);
    System.out.println("Now elements of queue2 = " + abq1);
    System.out.println("Rest element of queue1 = " + abq);
    Iterator it = abq1.iterator();
    System.out.println("Elements of queue2 using iterator = ");
    while (it.hasNext()) {
      System.out.println(it.next());
    }
    Object obj = abq1.peek();
    System.out.println("The head element of queue2 = " + obj);
    Object obj1 = abq1.poll();
    System.out.println("Elements of queue2 = " + abq1);
    System.out.println("The removed head element = " + obj1);
    int i1 = abq1.size();
   System.out.println("Size of queue2 = " + i1);
    int i2 = abq.size();
    System.out.println("Size of queue1 = " + i2);
  }
}

Output:

Elements of queue1= [1, 2, 3, 4, 5]

Elements of queue2 = [A, B, C, D, E, F]

Now elements of queue2 = [A, B, C, D, E, F, 1, 2, 3, 4]

Rest element of queue1 = [5]

Elements of queue2 using iterator =

A

B

C

D

E

F

1

2

3

4

The head element of queue2 = A

Elements of queue2 = [B, C, D, E, F, 1, 2, 3, 4]

The removed head element = A

Size of queue2 = 9

Size of queue1 = 1

Example 2 - Producer Consumer Problem

Lets look at the example:
This example has three components,
1. Producer Thread – This thread starts adding the data in to the Queue
2. Consumer Thread – This thread gets the data from the Queue whenever any data is added by the Producer.
3. Blocking Queue – This acts as an intermediate between the Producer and the Consumer thread. It gets the data or object from the producer thread and hands over to the consumer thread. 
ExecutorQueue.java
Let’s create an ExecutorQueue class which has the ArrayBlockingQueue object. Any data or object can be added and retrieved from the Array blocking Queue instance.

public class ExecutorQueue
{
public static BlockingQueue     queue   = new ArrayBlockingQueue(100);

/**
         * Method to add Data in to the Queue
         */
public static void addDataInQueue(Object obj)
{
queue.add(obj);
}

/**
         * Get the Data from the Queue
         */
public static Object getDataFromQueue() throws InterruptedException
{
return queue.take();
}
}

ConsumerThread.java
The below consumer Thread will wait in the Array blocking queue and retrieves the data as and when any object is added in the queue.

Note: In the Data Processing section, you can write your custom processing logic as per your requirements

public class ConsumerThread extends Thread
{
public void run()
{
System.out.println("\nConsumerThread started...");
boolean loop = true;
while (loop)
{
try
{
System.out.println("\nConsumerThread: Waiting to fetch data from Queue...");
String data = (String) ExecutorQueue.getDataFromQueue();
System.out.println("ConsumerThread: Got the data from Queue; Object = " + data);
System.out.println("ConsumerThread: Processing the data (" + data +")");

/*
                                 * Data Processing section:
                                 *
                                 * Note: Write your processing logic here based on the data retrieved
                                 */
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
}

Main method – ProducerApplication.java
The Producer is the triggering point to test Array blocking queue. It does the following operations,
1. It takes care of starting the Consumer Thread so that it will wait for any data in the Queue.
2. Soon after starting the Consumer thread, it will start adding data in to the blocking queue. Once the data is added, the Consumer thread will start retrieving the data using the queue.take() method.

public class ProducerApplication
{
public static void main(String[] args)
{

// Start the Processor thread so that it will wait for an Object in the Blocking Queue. As soon as an Object is
// added in the Queue, it will take and process the data
System.out.println("ProducerApplication: Starting the Processor Thread...\n");
ConsumerThread thread = new ConsumerThread();
thread.start();

ExecutorQueue.addDataInQueue("Object-1");
ExecutorQueue.addDataInQueue("Object-2");
}
}

Output

ProducerApplication: Starting the Processor Thread...


ConsumerThread started...

ConsumerThread: Waiting to fetch data from Queue...
ConsumerThread: Got the data from Queue; Object = Object-1
ConsumerThread: Processing the data (Object-1)

ConsumerThread: Waiting to fetch data from Queue...
ConsumerThread: Got the data from Queue; Object = Object-2
ConsumerThread: Processing the data (Object-2)

ConsumerThread: Waiting to fetch data from Queue...

Download the source

Source code can be downloaded from here.

Some good books on java concurrency

Java Concurrency in practise 
by Brian Goetz and others














Concurrent Programming in java
 Doug Lea

(Though old now, it is still good. The author is God of Concurrency in java but this book is written by a genius but not so good author)










Art of multiprocessing programming
- Maurice Herlihy












Effective Java 
  by Joshua Bloch

Though this book is mainly about good practices in java, still it has some good practices which will help you in writing concurrent code.

SingleThreadPool Example

This article will discuss about Thread pool that uses single thread to execute tasks. From Java 5.0+ one can get such pool from Executors using following method –

public static ExecutorService newSingleThreadExecutor()

It creates an Executor that uses a single worker thread operating off an unbounded queue. (Note however that if this single thread terminates due to a failure during execution prior to shutdown, a new one will take its place if needed to execute subsequent tasks.) Tasks are guaranteed to execute sequentially, and no more than one task will be active at any given time. Unlike the otherwise equivalent newFixedThreadPool(1) the returned executor is guaranteed not to be reconfigurable to use additional threads.

Example
Here is our worker thread:

public class WorkerThread implements Runnable {
    private int workerNumber;

    WorkerThread(int number) {
        workerNumber = number;
    }

    public void run() {
        for (int i=0;i<=100;i+=20) {
        // Perform some work ...
            System.out.println("Worker number: " + workerNumber
                + ", percent complete: " + i );
            try {
                Thread.sleep((int)(Math.random() * 1000));
            } catch (InterruptedException e) {
            }
        }
    }
}

This is our SingleThreadPoolDemo

public class SingleThreadPoolCemo {
   public static void main(String[] args) {
      ExecutorService svc = Executors.newSingleThreadExecutor();

      for(int i=0;i<4;i++){
         svc.submit(new Boogie(i));
      }
      svc.shutdown();

      System.out.println("Done work by workers");
   }

}

FixedThreadPool example

Creates a fixed-size thread pool. Here is the syntax:

public static ExecutorService 
               newFixedThreadPool(int nThreads)

Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue. At any point, at most nThreads threads will be active processing tasks. If additional tasks are submitted when all threads are active, they will wait in the queue until a thread is available. If any thread terminates due to a failure during execution prior to shutdown, a new one will take its place if needed to execute subsequent tasks. The threads in the pool will exist until it is explicitly shutdown.

Example
Here is a runnable task called WorkerThread (in a .java source file). The task performs some work and then periodically reports what percent of the work it has completed.

public class WorkerThread implements Runnable {
    private int workerNumber;

    WorkerThread(int number) {
        workerNumber = number;
    }

    public void run() {
        for (int i=0;i<=100;i+=20) {
        //Perform some work...
            System.out.format("Worker number: %d, percent complete: %d%n",
                workerNumber, i);
            try {
                Thread.sleep((int)(Math.random() * 1000));
            } catch (InterruptedException e) { }
        }
    }
}

In FixedThreadPoolDemo (in a .java source file), you can specify the number of worker threads to create and the size of the thread pool that will be used to run the threads. The following example uses a fixed thread pool so that you can observe the effect of running the program with fewer threads than tasks.

import java.util.concurrent.*;
public class FixedThreadPoolDemo {
    public static void main(String[] args) {
        int numWorkers = Integer.parseInt(args[0]);
        int threadPoolSize = Integer.parseInt(args[1]);
        ExecutorService tpes =
            Executors.newFixedThreadPool(threadPoolSize);
        WorkerThread[] workers = new WorkerThread[numWorkers];
        for (int i = 0; i < numWorkers; i++) {
            workers[i] = new WorkerThread(i);
            tpes.execute(workers[i]);
        }
        tpes.shutdown();
    }
}

Here is the result of running the test with 4 workers and a pool of 2 threads.

% java ThreadPoolTest 4 2
Worker number: 0, percent complete: 0
Worker number: 1, percent complete: 0
Worker number: 0, percent complete: 20
Worker number: 0, percent complete: 40
Worker number: 1, percent complete: 20
Worker number: 0, percent complete: 60
Worker number: 0, percent complete: 80
Worker number: 0, percent complete: 100
Worker number: 1, percent complete: 40
Worker number: 1, percent complete: 60
Worker number: 2, percent complete: 0
Worker number: 1, percent complete: 80
Worker number: 2, percent complete: 20
Worker number: 2, percent complete: 40
Worker number: 1, percent complete: 100
Worker number: 2, percent complete: 60
Worker number: 2, percent complete: 80
Worker number: 2, percent complete: 100
Worker number: 3, percent complete: 0
Worker number: 3, percent complete: 20
Worker number: 3, percent complete: 40
Worker number: 3, percent complete: 60
Worker number: 3, percent complete: 80
Worker number: 3, percent complete: 100

Notice how workers 0 and 1 are assigned to the two threads in the pool and that they alternately run to completion, then tasks 2 and 3 are assigned to the threads.

Like most of the other tasks in this chapter, WorkerThread implements the Runnable (in the API reference documentation) interface. Another way to create a task is to implement the Callable (in the API reference documentation) interface, as shown in the following example, CallableWorkerThread (in a .java source file). A Callable is more flexible than a Runnable because it can return a value and throw an exception. To implement a Callable, you provide the call method, which returns a value, in this case, an Integer that represents the task's number.

import java.util.concurrent.*;
public class CallableWorkerThread implements Callable<Integer> {
    private int workerNumber;

    CallableWorkerThread(int number) {
        workerNumber = number;
    }

    public Integer call() {
        for (int i = 0; i <= 100; i += 20) {
            //Perform some work...
            System.out.format("Worker number: %d, percent complete: %d%n",
                workerNumber, i);
            try {
                Thread.sleep((int)(Math.random() * 1000));
            } catch (InterruptedException e) {}
        }
        return(workerNumber);
    }
}

CachedThreadPool Example

This article will discuss about Thread pool that can reuse previously constructed threads when they are available. From Java 5.0+ one can get such pool from Executors using following method –

public static ExecutorService newCachedThreadPool();

Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available.So this means number of threads provided by cached thread pool is unbounded. So we can put it like this:
Creates an unbounded thread pool with automatic thread reclamation.

These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks. Calls to execute will reuse previously constructed threads if available. If no existing thread is available, a new thread will be created and added to the pool. Threads that have not been used for sixty seconds are terminated and removed from the cache. Thus, a pool that remains idle for long enough will not consume any resources. Note that pools with similar properties but different details (for example, timeout parameters) may be created using ThreadPoolExecutor constructors.

Example
Here is a runnable task called WorkerThread (in a .java source file). The task performs some work and then periodically reports what percent of the work it has completed.

public class WorkerThread implements Runnable {
    private int workerNumber;

    WorkerThread(int number) {
        workerNumber = number;
    }

    public void run() {
        for (int i=0;i<=100;i+=20) {
        //Perform some work...
            System.out.format("Worker number: %d, percent complete: %d%n",
                workerNumber, i);
            try {
                Thread.sleep((int)(Math.random() * 1000));
            } catch (InterruptedException e) { }
        }
    }
}

CachedThreadPoolDemo uses the CachedThreadPool executor service, which creates as many threads as needed but reuses previously constructed threads if available. You use the submit method to ask an executor service to run a Callable. This method returns a Future (in the API reference documentation) object, which gives you control over the task; you can use Future to retrieve the result of running the task, to monitor the task, and to cancel the task. For example, to gain access to the return result, simply call the method get.

public class CachedThreadPoolDemo {
    public static void main(String[] args) {
        int numWorkers = Integer.parseInt(args[0]);

        ExecutorService tpes =
            Executors.newCachedThreadPool();
        CallableWorkerThread workers[] = 
            new CallableWorkerThread[numWorkers];
        Future<Integer> futures[] = new Future[numWorkers];
        
        for (int i = 0; i < numWorkers; i++) {
            workers[i] = new CallableWorkerThread(i);
            futures[i]=tpes.submit(workers[i]);
        }

        for (int i = 0; i < numWorkers; i++) {
            try {
                System.out.format("Ending worker: %d%n",
                    futures[i].get());
            } catch (Exception e) {}
        }
    }
}

Output
Here is the output :

Worker number: 0, percent complete: 0
Worker number: 1, percent complete: 0
Worker number: 2, percent complete: 0
Worker number: 3, percent complete: 0
Worker number: 3, percent complete: 20
Worker number: 3, percent complete: 40
Worker number: 3, percent complete: 60
Worker number: 1, percent complete: 20
Worker number: 0, percent complete: 20
Worker number: 1, percent complete: 40
Worker number: 2, percent complete: 20
Worker number: 3, percent complete: 80
Worker number: 0, percent complete: 40
Worker number: 2, percent complete: 40
Worker number: 2, percent complete: 60
Worker number: 1, percent complete: 60
Worker number: 3, percent complete: 100
Worker number: 2, percent complete: 80
Worker number: 2, percent complete: 100
Worker number: 0, percent complete: 60
Worker number: 0, percent complete: 80
Worker number: 0, percent complete: 100
Worker number: 1, percent complete: 80
Ending worker: 0
Worker number: 1, percent complete: 100
Ending worker: 1
Ending worker: 2
Ending worker: 3

ScheduledThreadPool Example

This article will discuss about Thread pool that can schedule threads to run after a specified interval of time. From Java 5.0+ one can get such pool from Executors using following method –

public static ScheduledExecutorService 
                newScheduledThreadPool(int corePoolSize)

Creates a thread pool that can schedule commands to run after a given delay, or to execute periodically. The return type of this method (return type of thread pool) is ScheduledExecutorService. Some of the salient features of ScheduledExecutorService are –

1. Schedule a Callable or Runnable to run once with a fixed delay after submission
   
2. Schedule a Runnable to run periodically at a fixed rate
   
3. Schedule a Runnable to run periodically with a fixed delay between executions
   
4. Submission returns a ScheduledFutureTask handle which can be used to cancel the task
   
5. Like Timer, but supports pooling

Example

Lets look at the example. Suppose we have a thread (i.e.Runnable object) of type MyThread which replicates a typical application behaviour by sleeping for a user defined time duration. The end user configures the MyThread sleep duration by passing the interval as a constructor argument.

public class MyThread implements Runnable {

    private int delayTime = 0;

    public MyThread(int delayTime) {
        this.delayTime = delayTime;
    }

    public MyThread() {

    }

    @Override
    public void run() {
        Thread curThread = Thread.currentThread();
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd 'at' HH:mm:ss");
        System.out.println(df.format(new java.util.Date()) + " Starting thread " + curThread.getName());
        try {
            Thread.sleep(delayTime * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        System.out.println(this.toString());
        System.out.println(df.format(new java.util.Date()) + " Ending thread " + curThread.getName());
        
    }

    @Override
    public String toString() {
        return ConcurrencyUtils.getThreadInfo();
    }

}

ConcurrencyUtils.java
I have used concurrency utils to print thread id, name and pool information. From the output it is clear that all reuse the same thread. Now this is using ConcurrencyUtils to get Thread info:

public class ConcurrencyUtils {

    public static String getThreadInfo() {
        StringBuilder sb = new StringBuilder();
        Thread curThread = Thread.currentThread();
        sb.append("Thread Id: ");
        sb.append(curThread.getId());
        sb.append("\n");
        sb.append("Name: ");
        sb.append(curThread.getName());
        sb.append("\n");
        sb.append("Group: ");
        sb.append(curThread.getThreadGroup().getName());
        sb.append("\n");
        return sb.toString();
    }

    public static String getShortThreadInfo() {
        StringBuilder sb = new StringBuilder();
        Thread curThread = Thread.currentThread();
        sb.append("Thread Id: ");
        sb.append(curThread.getId());
        sb.append(", ");
        sb.append("Name: ");
        sb.append(curThread.getName());
        sb.append(", ");
        sb.append("Group: ");
        sb.append(curThread.getThreadGroup().getName());
        sb.append("\n");
        return sb.toString();
    }

    public static String retrieveCurrentDate(){
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd 'at' HH:mm:ss");
        return df.format(new java.util.Date());
    }

}

Similarly we have other thread which does the calculation of 2 numbers:

public class Calculator implements Callable<Integer> {

    public static final int OPERATION_ADD = 1;

    private int param1 = 0;
    private int param2 = 0;
    private int operation = 0;

    public Calculator(int param1, int param2, int operation) {
        this.param1 = param1;
        this.param2 = param2;
        this.operation = operation;
    }

    public int getParam1() {
        return param1;
    }

    public void setParam1(int param1) {
        this.param1 = param1;
    }

    public int getParam2() {
        return param2;
    }

    public void setParam2(int param2) {
        this.param2 = param2;
    }

    public int getOperation() {
        return operation;
    }

    public void setOperation(int operation) {
        this.operation = operation;
    }

    @Override
    public Integer call() throws Exception {
        int retValue = 0;
        switch (this.operation) {
        case 1:
            retValue = this.param1 + this.param2;
            break;
        // Additional cases can be similarly added for
        //other mathematical operations

        default:
            retValue = 0;
            break;
        }
        return Integer.valueOf(retValue) ;
    }
}

Demo.java
Finally we have main method, where we will call executor service to do our task. One more thing to note is the invocation of shutdown method on ExecutorService ,which is important else the Java process will not terminate.

public class ScheduledThreadPoolTest {

    public static void main(String[] args) {
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd 'at' HH:mm:ss");
        ScheduledExecutorService svc = Executors.newScheduledThreadPool(2);
        System.out.println(df.format(new java.util.Date()) + " Time Runnable");
        svc.schedule(new MyThread(), 2, TimeUnit.SECONDS);
        System.out.println(df.format(new java.util.Date()) + " Time Callable");
        ScheduledFuture<Integer> sf = svc.schedule(new Calculator(1, 3,
            Calculator.OPERATION_ADD), 5, TimeUnit.SECONDS);
        try {
            System.out.println("Waiting for value.");
            Integer val = sf.get();
            System.out.println(df.format(new java.util.Date()) + " Time Callable Retrieve");
            System.out.println("Computed Value: " + val);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        svc.shutdown();

    }
}

Download the source

You can download the source from here.

Java Locks : Re-entrant locks

Java concurrency library provides more control over synchronization than synchronized. Either we need to control types of access (read and write) separately, or it is cumbersome to use because either there is no obvious mutex or we need to maintain multiple mutexes. So doing it by synchronized will eat lot of time and will be buggy as well.
Thankfully, lock utility classes were added in Java 1.5 and make these problems easier to solve.

Java Reentrant Locks

Java has a few lock implementations in the java.util.concurrent.locks package.
The general classes of locks are nicely laid out as interfaces:

• Lock - the simplest case of a lock which can be acquired and released
• ReadWriteLock - a lock implementation that has both read and write lock types – multiple read locks can be held at a time unless the exclusive write lock is held

Java provides two implementations of these locks that we care about – both of which are reentrant (this just means a thread can reacquire the same lock multiple times without any issue).

• ReentrantLock - as you’d expect, a reentrant Lock implementation
• ReentrantReadWriteLock - a reentrant ReadWriteLock implementation

Extended capabilities with Reentrant locks

The ReentrantLock in the util.concurrent.locks package gives the developers some flexibility here. With ReentrantLock following are some of the options

1. tryLock() : With ReentrantLock, the thread can immediately return if it did not get the lock (if the lock is with some other thread).
2. tryLock(long timeout, TimeUnit unit):With ReentrantLock, the thread can wait for some duration to get hold of the lock. If it does not get the lock within some time, it will return.
3. lockInterruptibly() : With ReentrantLock, the thread waiting for the lock can be interrupted and cause it to come out with InterruptedException.

Now, let’s see some examples. So the general way to use re-entrant lock is like this :

final ReentrantLock _lock = new ReentrantLock();

private void method() throws InterruptedException
{
   //Trying to enter the critical section
   _lock.lock(); // will wait until this thread gets the lock
   try
   {
      // critical section
   }
   finally
   {
      //releasing the lock so that other threads can get notifies
      _lock.unlock(); 
   }       
}

Using optional “fairness” parameter with ReentrantLock
ReentrantLock accepts an optional “fairness” parameter in it’s constructor. Normally what happens is, whenever a thread releases the lock anyone of the waiting threads will get the chance to acquire that lock. But there is no predefined order or priority in the selection of the thread (at least from a programmers perspective).

But if we are specifying the fairness parameter as “true” while creating a new ReentrantLock object, it gives us the guaranty that the longest waiting thread will get the lock next. Sounds pretty nice right?

Use of “Condition” in ReentrantLock
Condition can be considered as a separation of monitor methods (wait(), notify() & notifyAll()). For each ReentrantLock we can define a set of conditions and based on that we can make the threads waiting & things like that.

import java.util.concurrent.locks.Condition;
final Condition _aCondition = _lock.newCondition();
private void method1() throws InterruptedException
{
   _lock.lock();
   try
   {
      while (condition 1)
      {
          // Waiting for the condition to be satisfied
          // Note: At this time, the thread will give up the lock
          // until the condition is satisfied. (Signaled by other threads)
          _aCondition.await();
      }
      // method body
   }
   finally
   {
      _lock.unlock();
   }       

}

private void method2() throws InterruptedException
{
   _lock.lock();
   try
   {
      doSomething();
      if (condition 2)
      {
          // Signaling other threads that the condition is satisfied
          // Wakes up any one of the waiting threads
          _aCondition.signal();

          // Wakes up all threads waiting for this condition
          _aCondition.signalAll();
      }       

      // method body
   }
   finally
   {
      _lock.unlock();
   }
}

Example

We will take the same example of counter again. We have already seen how to implement counter using synchronized keyword here. Here we will see how to implement using Reentrant locks:
Counter.java

package com.vaani.lock;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Counter {
    private int count;
    private Lock lock = new ReentrantLock();

    public int getNextValue() {
        try {
            lock.lock();
            count++;
        }
        finally {
            lock.unlock();
            return count;
        }

    }


}

Worker.java
Now worker threads starts on the counter object and start incrementing the value:

package com.vaani.lock;

public class Worker implements Runnable {
    private Counter counter;
    private boolean increment;
    private int count;

    public Worker(Counter counter, boolean increment, int count) {
        this.counter = counter;
        this.increment = increment;
        this.count = count;
    }

    public void run() {
        for (int i = 0; i < this.count; i++) {
            System.out.println(this.counter.getNextValue());


        }
    }
}

Now let's put worker's on work – as in demo:

package com.vaani.lock.demo;
import com.vaani.lock.*;
public class ReentrantLockDemo {
    public static void main(String[] args) throws Exception {
    Counter counter = new Counter();
    Thread t1 = new Thread(new Worker(counter, true, 10000));
    t1.start();
    Thread t2 = new Thread(new Worker(counter, false, 10000));
    t2.start();

    t1.join();
    t2.join();
    System.out.println("Final count: " + counter.getNextValue());
    }
}

Download the source

Source code above can be downloaded from here.