This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this article, we’ll focus on a the concept of Exchanger in the Java language.

Exchanger

The exchanger class provides a kind of point for two threads, where threads can exchange their objects with other threads.An Exchanger may be viewed as a bidirectional form of a SynchronousQueue. Exchangers may be useful in applications such as genetic algorithms and pipeline designs.
When a thread arrives at an exchange point, it is necessary to wait for another thread to arrive. When other partners come in threads, two threads go forward to exchange threads.

In below example two threads are passing Integer values to each others. Both the Thread will wait until they receive the Information

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

class FirstThread implements Runnable {
	private Exchanger<Integer> exchanger;

	public FirstThread(Exchanger<Integer> exchanger) {
		this.exchanger = exchanger;
	}

	@Override
	public void run() {
		try {
			System.out.println("Passing information form FirstThread");
			Integer exchange = exchanger.exchange(99);
			System.out.println("Information Sent From FirstThread");
			System.out.println("Received information from Second Thread." + exchange);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
class SecondThread implements Runnable {
	private Exchanger<Integer> exchanger;

	public SecondThread(Exchanger<Integer> exchanger) {
		this.exchanger = exchanger;
	}

	@Override
	public void run() {

		System.out.println("Receiving information from First Thread.");
		try {

			Integer exchange = exchanger.exchange(2);
			System.out.println("Received information from first Thread." + exchange);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}

}
public class ExchangerExample {
	public static void main(String[] args) {
		Exchanger<Integer> exchanger = new Exchanger<>();

		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
		newFixedThreadPool.submit(new FirstThread(exchanger));
		newFixedThreadPool.submit(new SecondThread(exchanger));
		newFixedThreadPool.shutdown();
	}
}

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this article, we’ll focus on a the concept of BlockingQueue in the Java language.

BlockingQueue

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

A BlockingQueue may be capacity bounded. At any given time it may have a remainingCapacity beyond which no additional elements can be put without blocking. A BlockingQueue without any intrinsic capacity constraints always reports a remaining capacity of Integer.MAX_VALUE.

BlockingQueue implementations are thread-safe. All queuing methods achieve their effects atomically using internal locks or other forms of concurrency control. However, the bulk Collection operations addAll, containsAll, retainAll and removeAll are not necessarily performed atomically unless specified otherwise in an implementation. So it is possible, for example, for addAll(c) to fail (throwing an exception) after adding only some of the elements in c.

let’s implement Producer and Consumer Problem using BlockingQueue

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

class BlockingQueueProducer implements Runnable
{
	private BlockingQueue<Integer> blockingQueue;
	private Random random;
	public BlockingQueueProducer(BlockingQueue<Integer> blockingQueue) {
		this.blockingQueue=blockingQueue;
		this.random = new Random();
		
	}
	@Override
	public void run() {
		while(true)
		{
			System.out.println("BlockingQueueProducer - Adding Elements");
			try {
				blockingQueue.put(random.nextInt(3000));
				System.out.println("Added Element : Current Size of Q " + blockingQueue.size());
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}		
	}	
}

class BlockingQueueConsumer implements Runnable
{
	BlockingQueue<Integer> blockingQueue;
	public BlockingQueueConsumer(BlockingQueue<Integer> blockingQueue) {
		this.blockingQueue=blockingQueue;
	}

	@Override
	public void run() {
		while (true) {

			try {
				System.out.println("BlockingQueueConsumer : iteam recived from Q " + blockingQueue.take() );
				Thread.sleep(2500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}	
}
public class BlockingQueueExample {

	public static void main(String[] args) {
		BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(5);
		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
		newFixedThreadPool.submit(new BlockingQueueConsumer(blockingQueue));
		newFixedThreadPool.submit(new BlockingQueueProducer(blockingQueue));
		newFixedThreadPool.shutdown();
	}
}

Key Points

put(E e)

Inserts the specified element into this queue, waiting if necessary for space to become available.

take()

Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.

poll()

Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an element to become available.

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this article, we’ll focus on a the concept of CyclicBarrier in the Java language.

CyclicBarrier

CyclicBarrier allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.

CyclicBarrier are Similar to CountDownLatch but CyclicBarrier provide some additional features like
Reseting CyclicBarrier & Supports an optional Runnable command that is run once per barrier point.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

class CyclicBarrierWorker implements Runnable
{
	private CyclicBarrier cyclicBarrier;
	private int workerId;
	private Random random;
	public CyclicBarrierWorker(CyclicBarrier cyclicBarrier ,int id) {
		this.cyclicBarrier=cyclicBarrier;
		this.workerId=id;
		this.random = new Random();
		
	}
	@Override
	public void run() {
		System.out.println("Starting worker ID " + this.workerId);
		try {
			Thread.sleep(random.nextInt(4000));
			System.out.println("Worker " + workerId + " Completed it's work, Reducing  count of cyclicBarrier " );
			cyclicBarrier.await();
		} catch (InterruptedException | BrokenBarrierException e) {
			e.printStackTrace();
		}	
	}	
}
public class CyclicBarrierExample {
		public static void main(String[] args) {
	
		CyclicBarrier cyclicBarrier = new CyclicBarrier(5, ()->System.out.println("Barrier point reach ::: All Task Completed"));
		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);
		IntStream.range(1,6)
				 .forEach(cnt->{newFixedThreadPool.submit(new CyclicBarrierWorker(cyclicBarrier, cnt));
				 });		
		System.out.println("All Task Submited");
		newFixedThreadPool.shutdown();		
	}
  }
}

Key Points

CyclicBarrier(int parties, Runnable barrierAction) :
Creates a new CyclicBarrier that will trip when the given number of parties (threads) are waiting upon it, and which will execute the given barrier action when the barrier is tripped, performed by the last thread entering the barrier.

getNumberWaiting()
Returns the number of parties currently waiting at the barrier.

reset
Resets the barrier to its initial state.

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this article, we’ll focus on a the concept of CountDownLatch in the Java language.

CountDownLatch

CountDownLatch enables a java thread to wait until other set of threads completes their tasks.

A CountDownLatch is initialized with a given count.
The await methods block until the current count reaches zero due to invocations of the countDown() method, after which all waiting threads are released and any subsequent invocations of await return immediately.
This is a one-shot phenomenon – the count cannot be reset. If you need a version that resets the count, consider using a CyclicBarrier.

e.g. Assume we have divided one task in 5 small independent task.
Now main thread should wait, till other 5 Threads finish there work.
In these scenarios CountDownLatch can be used.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class CountDownLatchWorker implements Runnable
{
	private CountDownLatch countDownLatch;
	private int workerId;

	public CountDownLatchWorker(CountDownLatch countDownLatch ,int workerId) {
		this.countDownLatch=countDownLatch;
		this.workerId=workerId;
	}
	@Override
	public void run() {
		System.out.println("Worker " + workerId + " Started" );
		try {
			Thread.sleep(workerId*1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("Worker " + workerId + " Completed it's work, Reducing  count of countDownLatch " );
		countDownLatch.countDown();		
	}	
}
public class CountDownLatchExample {
	public static void main(String[] args) throws InterruptedException {
		CountDownLatch countDownLatch = new CountDownLatch(5);
		ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
		IntStream.range(1, 6)
				  .forEach(cnt -> {
					  newCachedThreadPool.submit(new CountDownLatchWorker(countDownLatch, cnt));
				  });		
		System.out.println("Main Thread is wating for workers to finish!!!!!!");
		countDownLatch.await();
		System.out.println("Work of All Worker is Completed");
		newCachedThreadPool.shutdown();		
	}
}

Key Points

await Method
Causes the current thread to wait until the latch has counted down to zero, unless the thread is interrupted.
If the current count is zero then this method returns immediately.
If the current count is greater than zero then the current thread becomes disabled for thread scheduling purposes and lies dormant until one of two things happen:

countDown Method
Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
If the current count is greater than zero then it is decremented. If the new count is zero then all waiting threads are re-enabled for thread scheduling purposes.
If the current count equals zero then nothing happens.

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.
In this quick article, we’ll focus on a the concept of Callable & Future in the Java language.

Callable

The Callable interface represents a thread that can return the value. It’s the same as the Runnable interface but can return the value.
The callable interface can be used to Compute the value and return it to invoking thread.

Future

Futureis generic interface that represents value which will be returned by callable interface.
There are two methods to get actual value from Future.
get() : When this method is called, thread will wait for result indefinitely.
V get(long timeout, TimeUnit unit) : When this method is called, thread will wait for result only for specified time.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class CallableWorkerExample implements Callable<String>
{
	private String someReturnValue;
	public CallableWorkerExample(String someValue) {
		this.someReturnValue=someValue;	
	}
	@Override
	public String call() throws Exception {
	
		System.out.println("Working on call");
		Thread.sleep(3000);
		return this.someReturnValue;
	}	
}
public class CallableAndFuture {
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CallableWorkerExample worker= new CallableWorkerExample("Some Value");
		ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
		Future<String> submit = newSingleThreadExecutor.submit(worker);
		System.out.println("Task Submited");
		String returnValue = submit.get();
		System.out.println("Return value from Callable " +returnValue);		
	}
}

Key Points

Exception Handling
Runnable.run method does not throws exceptions but Callable.call method throws exception.
ExecutorService
ExecutorService.submit Submits a value-returning task for execution and returns a Future representing the pending results of the task.submitMethod can take Callable and Runnable task as input.But the execute Method Discussed in ExecutorServiceAndThreadPools only takes Runnable task as input.

This Article is part of Series of Articles on Java 8 Concurrency Tutorial.

In this quick article, we’ll focus on a the concept of ExecutorService Framework in the Java language.

ExecutorService is a framework simplifies the task of creating threads and managing thread life cycle. ExecutorService is an interface, We need its implementations in order to make any use of it.

ThreadPoolExecutor & ScheduledThreadPoolExecutor implementations are available in java concurrent package.

Creating an ExecutorService:

Executors factory methods are available for creating ExecutorService.

ExecutorService executorService1 = Executors.newSingleThreadExecutor();
ExecutorService executorService2 = Executors.newFixedThreadPool(10);
ExecutorService executorService3 = Executors.newCachedThreadPool();

newCachedThreadPool

newCachedThreadPool method creates an executor having an expandable thread pool.Whenever a thread is needed, pool returns a thread from cache and if not available, a new thread is created for a short time. When the timeout of thread is over, that thread is vanished.

In below example 10 Threds will run Simultaneously

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

 class MyWorker implements Runnable
{
	private int id;
	public MyWorker(int id) {
		this.id=id;
	}
	@Override
	public void run() {
		System.out.println("MyWorker id " + id + " IS Working" + "Start Time " + System.currentTimeMillis());
		try {
			Thread.sleep(5000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}		
	}	
}
public class ExecutorServiceExample {
	public static void main(String[] args) {
	ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
		IntStream.range(0, 10)
				 .forEach(cnt->newCachedThreadPool.execute(new MyWorker(cnt)));		
		newCachedThreadPool.shutdown();	
	}
}

newFixedThreadPool

newFixedThreadPool method Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue..

In below example 5 Threds will run Simultaneously. After Complition of task same 5 threds will be used for next 5 taks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

 class MyWorker implements Runnable
{
	private int id;
	public MyWorker(int id) {
		this.id=id;
	}
	@Override
	public void run() {
		System.out.println("MyWorker id " + id + " IS Working" + "Start Time " + System.currentTimeMillis());
		try {
			Thread.sleep(5000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}		
	}	
}
public class ExecutorServiceExample {
	public static void main(String[] args) {
	ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(5);
		IntStream.range(0, 10)
				 .forEach(cnt->newFixedThreadPool.execute(new MyWorker(cnt)));		
		newFixedThreadPool.shutdown();
	}
}

newSingleThreadExecutor

newSingleThreadExecutor method Creates an Executor that uses a single worker thread operating off an unbounded queue.

In below example 1 Thread will run Simultaneously. After Complition of task same threds will be used for next 10 taks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

 class MyWorker implements Runnable
{
	private int id;
	public MyWorker(int id) {
		this.id=id;
	}
	@Override
	public void run() {
		System.out.println("MyWorker id " + id + " IS Working" + "Start Time " + System.currentTimeMillis());
		try {
			Thread.sleep(5000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}		
	}	
}
public class ExecutorServiceExample {
	public static void main(String[] args) {
	ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
		IntStream.range(0, 10)
				 .forEach(cnt->newSingleThreadExecutor.execute(new MyWorker(cnt)));		
		newSingleThreadExecutor.shutdown();
	}
}

Key Points

shutdown()
An ExecutorService can be shut down, which will cause it to reject new tasks. Two different methods are provided for shutting down an ExecutorService.
The shutdown() method will allow previously submitted tasks to execute before terminating, while the shutdownNow() method prevents waiting tasks from starting and attempts to stop currently executing tasks.

execute()
Executes the given command at some time in the future. The command may execute in a new thread, in a pooled thread, or in the calling thread, at the discretion of the Executor implementation.
execute method only takes runnable task.

Semaphores are a really simple concept, invented by the famous Dutch computer scientist Edsger Dijkstra.

Basically a semaphore is a counter (integer) that allows a thread to get into a critical region.
What the counter is counting are permits that allow access to the shared resource. Thus, to access the resource, a thread must be granted a permit from the semaphore.

If the value of the counter is greater than 0 then thread get the permit otherwise waits for the permit.
Once thread leaves the critical region increments the counter so that other thread can access the critical section.

Most of the time we use semaphores to limit the number of concurrent threads accessing a specific resource.

Example
let consider, We want to limit connections to some resources to some max limit.Similar to connection pool.
In below example 10 threads are trying to get connection at same time.But we should not allow more than 5 connections

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

package com.nirajsonawane;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.stream.IntStream;

class ConnectionPool {
	
	private Semaphore connectionSemaphore;

	public ConnectionPool(int poolsize) {
		this.connectionSemaphore = new Semaphore(poolsize);
	}

	public void getConnectionFromPool() {
		if (connectionSemaphore.availablePermits() > 0) {
			connectionSemaphore.tryAcquire();
			System.out.println("Get the connection");
		} else {
			System.out.println("Max active connection limit reach!! try again");
		}
	}
    public void releaseConnection()
	{
		connectionSemaphore.release();
	}
}
class ConnectionService implements Runnable {

	private ConnectionPool connectionPool;
	public ConnectionService(ConnectionPool connectionPool) {
		this.connectionPool = connectionPool;
	}

	@Override
	public void run() {
		connectionPool.getConnectionFromPool();
	}
}
public class Semaphores {

	public static void main(String[] args) {

		ExecutorService executorService = Executors.newFixedThreadPool(10);
		ConnectionPool connectionPool = new ConnectionPool(5);
		ConnectionService service = new ConnectionService(connectionPool);
		IntStream.range(0, 10)
				.forEach((cnt) -> executorService.execute(service));
	}
}
}

Out of 10 threds only 5 was able to get the connection.

Key Points

1. tryAcquire()– Return true if a permit is available immediately and acquire it otherwise return false,
2. acquire()- Acquires a permit and blocking until one is available.
3. release() – Release a permit
4. availablePermits() – Return number of current permits available

This is Sixth Article in Series of Articles on Java 8 Concurrency Tutorial.

In this article, we will look at one of the most basic methods of Java-thread synchronization.

Object Class in Java has three final methods to allow threads to communicate about the lock status of the threads.

These methods are wait(), notify() and notifyAll().

wait()

Causes the current thread to wait until another thread invokes the notify() method or the notifyAll() method for this object.

notify()

Wakes up a single thread that is waiting on this object’s monitor.

notifyAll()

Wakes up all threads that are waiting on this object’s monitor.

Let’s try to implement Producer & Consumer problem using wait(),notify()& notifyAll();

Producer will add number in List. List can have maximum 5 numbers . Consumer will remove elements from list until it becomes empty.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

 
 package com.nirajsonawane;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

class Processor {

	private List<Integer> list = new ArrayList<>();
	private Random random = new Random();

	public synchronized void producer() {

		while (true) {
			try {
				Thread.sleep(1000);
				if (list.size() == 5) {
					System.out.println("List is full Notifying Consumer & Releasing Lock");
					notifyAll();
					wait();
				}
				while (list.size() < 5) {
					System.out.println("Adding items");
					list.add(random.nextInt());
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public synchronized void consumner() {
		while (true) {

			try {
				Thread.sleep(1000);
				if (list.isEmpty()) {

					System.out.println("List is Empty :Notifying Publisher & Releasing Lock");
					notifyAll();
					wait();
				} else {
					System.out.println("Size of list " + list.size() + " Removed Number is " + list.remove(0));
				}

			} catch (Exception e) {
				// TODO: handle exception
			}
		}
	}
}

public class WaitNotifyNotifyAllExample {

	public static void main(String[] args) throws InterruptedException {
		Processor p = new Processor();
		Thread t1 = new Thread(() -> p.producer());
		Thread t2 = new Thread(() -> p.consumner());
		t1.start();
		t2.start();
		t1.join();
		t2.join();
	}

}

This is Fifth Article in Series of Articles on Java 8 Concurrency Tutorial.

In this quick article, we’ll focus on a the concept of volatile keyword in the Java language.

Every read of a volatile variable will be read from the RAM so from the main memory and not from cache. Usually variables are cached for performance reasons.

Using volatile variables reduces the risk of memory consistency errors, because any write to a volatile variable establishes a happens-before relationship with subsequent reads of that same variable. This means that changes to a volatile variable are always visible to other thread.

1

private static volatile int COUNT = 0;

This is Fourth Article in Series of Articles on Java 8 Concurrency Tutorial.

Intrinsic Locks

Synchronization is built around an internal entity known as the intrinsic lock or monitor lock. Intrinsic locks play a role in both aspects of synchronization: enforcing exclusive access to an object’s state and establishing happens-before relationships that are essential to visibility.

Every object has an intrinsic lock associated with it. By convention, a thread that needs exclusive and consistent access to an object’s fields has to acquire the object’s intrinsic lock before accessing them, and then release the intrinsic lock when it’s done with them.

As long as a thread owns an intrinsic lock, no other thread can acquire the same lock. The other thread will block when it attempts to acquire the lock.

Locks In Synchronized Methods

We have discussed Synchronized Methods in previous Article Synchronization.
When a thread invokes a synchronized method, it automatically acquires the intrinsic lock for that method’s object and releases it when the method returns. The lock release occurs even if the return was caused by an uncaught exception.

Locks In Synchronized Static Methods

As a static method is associated with a class, not an object. In this case, the thread acquires the intrinsic lock for the Class object associated with the class. Thus access to class’s static fields is controlled by a lock that’s disti