concurrent package

Callable and Future Interface

June 20, 2017 By j2eereference Leave a Comment

Callable and Future Interface :

Callable and Future interface has been introduced in JDK 1.5 under java.util.cocurrent package.

Callable interface is used to execute task and is similar to Runnable interface but can return value to caller and is able to throw checked Exception as well .Callable interface has call() method to execute task, computing result and returning result or throws an exception if unable to execute.

Callable interface is used by Executor framework, which contains pool of threads called worker threads then will call submit method of ExecutorService which is used to execute the Callable tasks by available threads in the pool.

When we submit callable task then returned result is an instance of Future object. This Future object is used to find out the status of callable task ,get the result of task computation or cancel the task.

Example using future and Callable :

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class CallableDemo {
  private static final int WORKER_THREADS = 10;
  public static void main(String[] args) throws InterruptedException,ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(WORKER_THREADS);
        Future<Integer> futureAdd = executor.submit(new AdditionCallable(10));
        System.out.println("Computation result returned after submitting Callable task "
         + futureAdd.get());
        executor.shutdown();
  }
}

class AdditionCallable implements Callable&lt;Integer&gt; {
  Integer n;
  AdditionCallable(Integer n) {
     this.n = n;
  }

  @Override
  public Integer call() throws Exception {
   Integer sum = 0;
   for (int i = 0; i <= n; i++) {
       System.out.println("value added in sum is " + i);
       sum += i;
       System.out.println("Updated sum is " + sum);
   }
   return sum;
  }
}

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class CallableDemo {

private static final int WORKER_THREADS = 10;

public static void main(String[] args) throws InterruptedException,ExecutionException {

ExecutorService executor = Executors.newFixedThreadPool(WORKER_THREADS);

Future<Integer> futureAdd = executor.submit(new AdditionCallable(10));

System.out.println("Computation result returned after submitting Callable task "

+ futureAdd.get());

executor.shutdown();

}

class AdditionCallable implements Callable<Integer> {

Integer n;

AdditionCallable(Integer n) {

this.n = n;

}

@Override

public Integer call() throws Exception {

Integer sum = 0;

for (int i = 0; i <= n; i++) {

System.out.println("value added in sum is " + i);

sum += i;

System.out.println("Updated sum is " + sum);

}

return sum;

}

OutPut:

value added in sum is 0
Updated sum is 0
value added in sum is 1
Updated sum is 1
value added in sum is 2
Updated sum is 3
value added in sum is 3
Updated sum is 6
value added in sum is 4
Updated sum is 10
value added in sum is 5
Updated sum is 15
value added in sum is 6
Updated sum is 21
value added in sum is 7
Updated sum is 28
value added in sum is 8
Updated sum is 36
value added in sum is 9
Updated sum is 45
value added in sum is 10
Updated sum is 55
Computation result returned after submitting Callable task 55

Methods of Callable and Future :
1) V call() throws Exception: This method returns result of executable task or throws an exception .
2) Get() :this method is used to get the result of callable task.It is a blocking call which means that if task is not finishes then it will wait until it is finished.
3) get(long timeout, TimeUnit): this method will wait at most the time specified in the method for the task to complete and give results if available.
4) Boolean Cancel() : this method will return true if attempt to cancel the task is successful. If task is already completed or cancelled then attempt will fail.
5) Boolean isCancelled(): this method will return true if task is cancelled before it gets complete.
6) Boolean isDone() : this method will return true if task compeltes.Completion of task can be exception,normal or cancelled.

What is exchanger in concurrency?

June 12, 2017 By j2eereference Leave a Comment

Exchanger :

Exchanger is introduced in jdk1.5 with classes like CountDownLatch,CyclicBarrier and Semaphores.

This class is used to exchange data between two threads. An Exchanger waits until both the threads call its exchange() method and when this method is invoked, the exchanger exchanges data between two threads. It uses method exchange for exchanging data .

It has two version of exchange method given below :

V exchange(V x): This method will wait until other thread calls exchange method or interrupts it. When other thread calls exchange method then waiting thread resumed and data exchanging is done between two threads.
V exchange(V x, long timeout, TimeUnit unit) : This method will wait until other thread calls exchange method or interrupts it or specified time elapses. . When other thread calls exchange method then waiting thread resumed and data exchanging is done between two threads.

Example to demonstrate Exchanger :

import java.util.concurrent.Exchanger;
class ProducerBuffer implements Runnable {
  Exchanger<String> exchanger;
  String message;
  ProducerBuffer(Exchanger<String> exchanger) {
     message = new String();
     this.exchanger = exchanger;
  }

 @Override
 public void run() {
    for (int i = 0; i < 10; i++) {
      message += i;
      System.out.println("Producer produced message: " + message);
      try {
       message = exchanger.exchange(message);
     } catch (InterruptedException e) {
       e.printStackTrace();
     }
   }
 }
}

class ConsumerBuffer extends Thread {
  ProducerBuffer prod;
  Exchanger<String> exchanger;
  String message;
  ConsumerBuffer(Exchanger<String> exchanger) {
     this.exchanger = exchanger;
  }

  public void run() {
   for (int i = 0; i < 10; i++) {
     try {
       message = exchanger.exchange(new String());
       System.out.println("Consumer consumed message: " + message);
     } catch (InterruptedException e) {
       e.printStackTrace();
    }
   }
  }
}
public class ExchangerDemo {
  public static void main(String[] args) {
   Exchanger<String> exchanger = new Exchanger<String>();
   ProducerBuffer prodBuffer = new ProducerBuffer(exchanger);
   ConsumerBuffer consBuffer = new ConsumerBuffer(exchanger);
   Thread prodThread = new Thread(prodBuffer, "prodThread");
   Thread consThread = new Thread(consBuffer, "consThread");
   prodThread.start();
   consThread.start();
  }
}

import java.util.concurrent.Exchanger;

class ProducerBuffer implements Runnable {

Exchanger<String> exchanger;

String message;

ProducerBuffer(Exchanger<String> exchanger) {

message = new String();

this.exchanger = exchanger;

}

@Override

public void run() {

for (int i = 0; i < 10; i++) {

message += i;

System.out.println("Producer produced message: " + message);

try {

message = exchanger.exchange(message);

} catch (InterruptedException e) {

e.printStackTrace();

}

class ConsumerBuffer extends Thread {

ProducerBuffer prod;

Exchanger<String> exchanger;

String message;

ConsumerBuffer(Exchanger<String> exchanger) {

this.exchanger = exchanger;

}

public void run() {

for (int i = 0; i < 10; i++) {

try {

message = exchanger.exchange(new String());

System.out.println("Consumer consumed message: " + message);

} catch (InterruptedException e) {

e.printStackTrace();

}

public class ExchangerDemo {

public static void main(String[] args) {

Exchanger<String> exchanger = new Exchanger<String>();

ProducerBuffer prodBuffer = new ProducerBuffer(exchanger);

ConsumerBuffer consBuffer = new ConsumerBuffer(exchanger);

Thread prodThread = new Thread(prodBuffer, "prodThread");

Thread consThread = new Thread(consBuffer, "consThread");

prodThread.start();

consThread.start();

}

OutPut:

Producer produced message: 0
Producer produced message: 1
Consumer consumed message: 0
Consumer consumed message: 1
Producer produced message: 2
Consumer consumed message: 2
Producer produced message: 3
Consumer consumed message: 3
Producer produced message: 4
Consumer consumed message: 4
Producer produced message: 5
Consumer consumed message: 5
Producer produced message: 6
Consumer consumed message: 6
Producer produced message: 7
Consumer consumed message: 7
Producer produced message: 8
Consumer consumed message: 8
Producer produced message: 9
Consumer consumed message: 9

Real Time usecase:
Exchanger is mainly used which implements Producer consumer problem as booth threads exchange data between each other

What is ReentrantLock?

May 4, 2017 By j2eereference Leave a Comment

ReentrantLock is an implementation of Lock interface and comes under java.util.concurrent.locks. ReentrantLock concept is same as traditional synchronized methods but it has more capability like fairness approach,Lock interruptibly ,tryLock and other features which we will discuss under important method of ReentrantLock section.

Constructors:

ReentrantLock() : this constructor is used to create instance of ReentrantLock which means fair value is false .

ReentrantLock(boolean fair): this construct will create instance of ReentrantLock which accept Boolean value as parameter either true or false.

If boolean variable fair is true it means that thread which is waiting from long will get the lock first

If boolean variable flag is false then it means any thread can get the lock .

Example:

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

public class ReentrantLockDemo {
  public static void main(String[] args) {
   Lock lock = new ReentrantLock();
   WorkerThread workerThread = new WorkerThread(lock);
   new Thread(workerThread, "WorkerThread1").start();
   new Thread(workerThread, "WorkerThread2").start();
   new Thread(workerThread, "WorkerThread3").start();
 }
}

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockDemo {

public static void main(String[] args) {

Lock lock = new ReentrantLock();

WorkerThread workerThread = new WorkerThread(lock);

new Thread(workerThread, "WorkerThread1").start();

new Thread(workerThread, "WorkerThread2").start();

new Thread(workerThread, "WorkerThread3").start();

}

class WorkerThread implements Runnable {
   Lock lock;
   public WorkerThread(Lock lock) {
   this.lock = lock;
 }

  public void run() {
   lock.lock();
   System.out.println("Lock has been acquired by "+ Thread.currentThread().getName());
   try {
    Thread.sleep(1000);
   } catch (InterruptedException e) {
     e.printStackTrace();
   }
   System.out.println("Lock has been released by "+ Thread.currentThread().getName());
   lock.unlock();
  }
}

class WorkerThread implements Runnable {

Lock lock;

public WorkerThread(Lock lock) {

this.lock = lock;

}

public void run() {

lock.lock();

System.out.println("Lock has been acquired by "+ Thread.currentThread().getName());

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("Lock has been released by "+ Thread.currentThread().getName());

lock.unlock();

}

Output:
Lock has been acquired by WorkerThread1
Lock has been released by WorkerThread1
Lock has been acquired by WorkerThread2
Lock has been released by WorkerThread2
Lock has been acquired by WorkerThread3
Lock has been released by WorkerThread3

Important Methods:

1) void lock() : This method will cause current thread to hold the lock if it is not acquired by other thread otherwise current thread will wait until acquired lock gets released by other thread.
One more scenario is that lock hold count would increased by 1 in case lock is already held by current thread
2)void unlock(): If current thread calls unlock method then lock is released if lock hold count is zero and if it greater than 0 then lock hold count get decremented by 1.
3)boolean tryLock(): This method will acquires lock and return true if lock is not held by any thread otherwise returns false.
4)boolean tryLock(long timeout, TimeUnit unit):This method will wait for the lock to become available until time mentioned in the method timeouts.
5)void lockInterruptibly():This method allows the thread to immediately react to the interrupt signal sent by other thread and throw interrupted exception.
6)getHoldCount() :This method will give you hold count on this lock by the current thread.
7)isHeldByCurrentThread() :This method will check whether current thread holds the lock .

Why AtomicInteger class

April 27, 2017 By j2eereference 1 Comment

AtomicInteger class belongs to java.util.concurrent.atmoic package which contain many classes like AtomicInteger,AtomicLong and AtomicBoolean.AtomicInteger is a class that provides you an int value which can be read and write atomically by many threads without interrupting each other

An atomic operation is one which gives guarantee that operation will not get interrupted in between by the thread scheduler .

Atomic classes internally using CAS (compare and swap) algorithm.

How it works in Multi threaded environment ?

Lets see how the below program works in the multi threaded environment with primitive int data type

class Count 
{
	public static int countIntValue=0;
}
public class RunnableAtomicIntegerDemo extends Thread{
	@Override
	public void run() { 
		Count.countIntValue++;
		System.out.print( Count.countIntValue+"  ");
	}
	public static void main(String[] args) {
		for(int i=0;i<=15;i++){
			new Thread(new RunnableAtomicIntegerDemo()).start();
		}
	}
}

class Count

{

public static int countIntValue=0;

}

public class RunnableAtomicIntegerDemo extends Thread{

@Override

public void run() {

Count.countIntValue++;

System.out.print( Count.countIntValue+" ");

}

public static void main(String[] args) {

for(int i=0;i<=15;i++){

new Thread(new RunnableAtomicIntegerDemo()).start();

}

Output :

5 9 10 8 7 7 7 7 7 6 11 12 13 14 16 16

Here you can notice that we have duplicate values because of the multiple thread access. This can be avoided by the following program with the use of Synchronized block.

class Count 
{
	public static int countIntValue=0;
}
public class RunnableAtomicIntegerDemo extends Thread{
	@Override
	public void run() { 
		synchronized (Count.class) {
			Count.countIntValue++;
			System.out.print( Count.countIntValue+"  ");
		}
	}
	public static void main(String[] args) {
		for(int i=0;i<=15;i++){
			new Thread(new RunnableAtomicIntegerDemo()).start();
		}
	}
}

class Count

{

public static int countIntValue=0;

}

public class RunnableAtomicIntegerDemo extends Thread{

@Override

public void run() {

synchronized (Count.class) {

Count.countIntValue++;

System.out.print( Count.countIntValue+" ");

}

public static void main(String[] args) {

for(int i=0;i<=15;i++){

new Thread(new RunnableAtomicIntegerDemo()).start();

}

Output:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

Here we don’t have any duplicate values as we used Synchronized block.

Here comes the benefit of using Atomic classes. Lets see how the above program can be implemented easily by using AtomicInteger class.

import java.util.concurrent.atomic.AtomicInteger;

class Count 
{
	public static AtomicInteger countValue=new AtomicInteger(0);
}
public class RunnableAtomicIntegerDemo extends Thread{
	@Override
	public void run() { 
			System.out.print("   "+Count.countValue.incrementAndGet());
	}
	public static void main(String[] args) {
		for(int i=0;i<=15;i++){
			new Thread(new RunnableAtomicIntegerDemo()).start();
		}
	}
}

import java.util.concurrent.atomic.AtomicInteger;

class Count

{

public static AtomicInteger countValue=new AtomicInteger(0);

}

public class RunnableAtomicIntegerDemo extends Thread{

@Override

public void run() {

System.out.print(" "+Count.countValue.incrementAndGet());

}

public static void main(String[] args) {

for(int i=0;i<=15;i++){

new Thread(new RunnableAtomicIntegerDemo()).start();

}

Output:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

Program to demonstrate different methods :

import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerDemo {
	public static void main(String[] args) {
		AtomicInteger atomicInt = new AtomicInteger(5);
		System.out.println("Value returned by get() is " + atomicInt.get());
		atomicInt.set(7);
		System.out.println("Value returned after calling set(7) is "+ atomicInt.get());
		System.out.println("Value returned on calling getAndSet(10) is "+ atomicInt.getAndSet(10));
		System.out.println("Value returned after calling getAndSet(10) is "+ atomicInt.get());
		atomicInt.addAndGet(5);
		System.out.println("Value returned after calling addAndGet(5) is "+ atomicInt.get());
		atomicInt.incrementAndGet();
		System.out.println("Value returned after calling incrementAndGet() is "+ atomicInt.get());
		atomicInt.decrementAndGet();
		System.out.println("Value returned after calling decrementAndGet() is "+ atomicInt.get());
		atomicInt.compareAndSet(15, 20);
		System.out.println("Value returned after calling compareAndSet(15,20) is "+ atomicInt.get());
	}
}

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerDemo {

public static void main(String[] args) {

AtomicInteger atomicInt = new AtomicInteger(5);

System.out.println("Value returned by get() is " + atomicInt.get());

atomicInt.set(7);

System.out.println("Value returned after calling set(7) is "+ atomicInt.get());

System.out.println("Value returned on calling getAndSet(10) is "+ atomicInt.getAndSet(10));

System.out.println("Value returned after calling getAndSet(10) is "+ atomicInt.get());

atomicInt.addAndGet(5);

System.out.println("Value returned after calling addAndGet(5) is "+ atomicInt.get());

atomicInt.incrementAndGet();

System.out.println("Value returned after calling incrementAndGet() is "+ atomicInt.get());

atomicInt.decrementAndGet();

System.out.println("Value returned after calling decrementAndGet() is "+ atomicInt.get());

atomicInt.compareAndSet(15, 20);

System.out.println("Value returned after calling compareAndSet(15,20) is "+ atomicInt.get());

}

OutPut:
Value returned by get() is 5
Value returned after calling set(7) is 7
Value returned on calling getAndSet(10) is 7
Value returned after calling getAndSet(10) is 10
Value returned after calling addAndGet(5) is 15
Value returned after calling incrementAndGet() is 16
Value returned after calling decrementAndGet() is 15
Value returned after calling compareAndSet(15,20) is 20

Important methods :

1)int get() : This method will return current value of AtomicInteger
2)void set(int value) : This method will set the given value in AtomicInteger
3)int getAndSet(int newValue) : This method will set the given value and then returns the old value defined in AtomicInteger constructor.
4)int addAndGet(int value) : Method will add the given value in the value defined in AtomicInteger Constructor and returns the updated value.
5)int incrementAndGet() : This method will increment the value given in AtomicInteger by 1 and return the updated value.
6)boolean compareAndSet(int expect, int update) : Method will Compare the value given in Atomic class constructor with the expect value given in the method and if both are equal then set to update value and returns true.
7)int decrementAndGet():This method will decrements the value given in AtomicInteger by 1 and return the updated value.

Advantages of Atomic classes:

It will be very useful when you are working in multithreading environment and required to perform thread safe operations on integer without taking any lock(synchronized method)
Operations are fast and more readable than primitive synchronized methods.
It provides many operations like increment and decrement on integer variable which is shared between many threads and hence minimizing the need of complex multithreading.

What is CyclicBarrier?

April 25, 2017 By j2eereference 4 Comments

CyclicBarrier is another concurrency utility introduced in JDK 1.5 , Whis is a java synchronizer that allows two or more threads to wait for each other at a common point known as the barrier before it starts processing.
When all threads have reached barrier point then all waiting threads are released, and if any Runnable action is defined then it will be triggered as well.

CyclicBarrier’s constructor

There are two constructors with the below parameters.

(int parties) : Here parties is the number of threads need to wait at barrier.
(int parties, Runnable barrierAction) : The second parameter is the barrier action which will get executed after all threads have reached common barrier point.

Important methods:

await();This method is similar to countdown() method of CountDownLatch .This method is called on the CyclicBarrier object in the thread which is to be paused. If there are 4 threads that must pause at barrier point then await() must be called in the run() of those threads when the last thread (4) calls the await() method, after that every thread should wake up and start processing.
await(long time,TimeUnit units):This method is same as the above method but the thread here is paused for a given time If barrier is broken before a thread calls await() then this method will throw BrokenBarrierException.

reset(); This method resets the barrier to the initial state.

How it works ?

Create instance of CyclicBarrier and initialized it with 4(working threads), this is actually number of threads need to wait at barrier. The barrier will not be broken until all threads are arrived. A thread is said to be arrived when it calls await() method.

In below program, we will use four worker threads and one main thread, which is running your main method. We have given each thread a different name, starting from Thread-1 to Thread-4 just to have a better understanding. Now pass same instance of the cyclic barrier to each thread. In run method implementation we define sleep method that causes each thread to sleep for some seconds and then call await() method on the barrier.

Example:

In below program, we will use four worker threads and one main thread, which is running your main method

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

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

		CyclicBarrier cyclicBarrier=new CyclicBarrier(4 ,new RunnableAction());
		System.out.println("Created CyclicBarrier with 4 threads and Runnable Action will be triggered when all "
				+ "4 threads will reach at common barrier point ");
		// Creating and starting 4 threads
		for(int i=1;i<=4;i++){
			new Thread(new CyclicRunnable(cyclicBarrier),"Thread-"+i).start();
		}

	}

}

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {

public static void main(String[] args) {

CyclicBarrier cyclicBarrier=new CyclicBarrier(4 ,new RunnableAction());

System.out.println("Created CyclicBarrier with 4 threads and Runnable Action will be triggered when all "

+ "4 threads will reach at common barrier point ");

// Creating and starting 4 threads

for(int i=1;i<=4;i++){

new Thread(new CyclicRunnable(cyclicBarrier),"Thread-"+i).start();

}

class RunnableAction implements Runnable{

	@Override
	public void run()
	{
		System.out.println("Runnable Action will be triggered when all the threads reach barrier point");
	}
}

class RunnableAction implements Runnable{

@Override

public void run()

{

System.out.println("Runnable Action will be triggered when all the threads reach barrier point");

}

class CyclicRunnable implements Runnable{

	private CyclicBarrier cyclicBarrier;

	CyclicRunnable(CyclicBarrier cyclicBarrier){
		this.cyclicBarrier=cyclicBarrier;
	}

	@Override
	public void run() {
		System.out.println(Thread.currentThread().getName() +" is arrived at common barrier point");
		try {
			Thread.sleep(1000);
			cyclicBarrier.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (BrokenBarrierException e) {
			e.printStackTrace();
		}

		System.out.println(Thread.currentThread().getName() + " has been released for further processing");
	}
}

class CyclicRunnable implements Runnable{

private CyclicBarrier cyclicBarrier;

CyclicRunnable(CyclicBarrier cyclicBarrier){

this.cyclicBarrier=cyclicBarrier;

}

@Override

public void run() {

System.out.println(Thread.currentThread().getName() +" is arrived at common barrier point");

try {

Thread.sleep(1000);

cyclicBarrier.await();

} catch (InterruptedException e) {

e.printStackTrace();

} catch (BrokenBarrierException e) {

e.printStackTrace();

}

System.out.println(Thread.currentThread().getName() + " has been released for further processing");

}

Output:
Created CyclicBarrier with 4 threads and Runnable Action will be triggered when all 4 threads will reach at common barrier point
Thread-1 is arrived at common barrier point
Thread-3 is arrived at common barrier point
Thread-2 is arrived at common barrier point
Thread-4 is arrived at common barrier point
Runnable Action will be triggered when all the threads reach barrier point
Thread-1 has been released for further processing
Thread-4 has been released for further processing
Thread-2 has been released for further processing
Thread-3 has been released for further processing

Realtime Use cases :

Multiplayer video game, game is started only when all players joined.
If you are going to a picnic, and you need to first meet at some common point from where you all will start your journey.
It can be used where big task is broken down into smaller tasks and to complete the job you need output from individual small task

Differences between CountDownLatch and CyclicBarrier:

CyclicBarrier has reset method which is not present in CountDownLatch,so CyclicBarrier can be used again once count becomes 0
CyclicBarrier has constructor that defines Runnable event,so it can be use to trigger Runnable event once counter becomes 0 but CountDownLatch cant not do this. Refer countDownLatch

concurrent package

Related Posts

Related Posts

Related Posts

Related Posts

Footer