vogella.de

The base means for concurrency are java.lang.Threads. Each thread requires a java.lang.Runnable which contains in the method run the unit of work which must be performed.

Think of runnable as the task to perform. The thread is the worker who is doing this task.

For example this is a simple task (Runnable) which counts the sum of a given range of numbers.

			
package de.vogella.concurrency.threads;

/**
 * MyRunnable will count the sum of the number from 1 to the parameter
 * countUntil and then write the result to the console.
 * <p>
 * MyRunnable is the task which will be performed
 * 
 * @author Lars Vogel
 * 
 */
public class MyRunnable implements Runnable {
	private final long countUntil;

	MyRunnable(long countUntil) {
		this.countUntil = countUntil;
	}

	@Override
	public void run() {
		long sum = 0;
		for (long i = 1; i < countUntil; i++) {
			sum += i;
		}
		System.out.println(sum);
	}
}

		

To perform these tasks (Runnables) you need threads. The following will create threads, assign runnable to these threads and wait until they are finished.

			
package de.vogella.concurrency.threads;

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

public class Main {

	public static void main(String[] args) {
		// We will store the threads so that we can check if they are done
		List<Thread> threads = new ArrayList<Thread>();
		// We will create 500 threads
		for (int i = 0; i < 500; i++) {
			Runnable task = new MyRunnable(10000000L + i);
			Thread worker = new Thread(task);
			// We can set the name of the thread
			worker.setName(String.valueOf(i));
			// Start the thread, never call method run() direct
			worker.start();
			// Remember the thread for later usage
			threads.add(worker);
		}
		int running = 0;
		do {
			running = 0;
			for (Thread thread : threads) {
				if (thread.isAlive()) {
					running++;
				}
			}
			System.out.println("We have " + running + " running threads. ");
		} while (running > 0);

	}
}

		

Threads were the first approach in Java to support concurrency. The usage of the threads has the following disadvantages.

The java.util.concurrent package offers improved support for concurrency compared to threads. The java.util.concurrent package helps solving several of the issues with threads.

Thread pools manage a pool of worker threads. The thread pools contains a work queue which holds tasks waiting to get executed.

A thread pool can be described as a collection of runnables (work queue) and a connections of running threads. These threads are constantly running and are checking the work query for new work. If there is new work to be done they execute this Runnable. The Thread class itself provides a method, e.g. execute(Runnable r) to add runnables to the work queue.

The Executor framework provides example implementation of the java.util.concurrent.Executor interface, e.g. Executors.newFixedThreadPool(int n) which will create n worker threads. The ExecutorService adds lifecycle methods to the Executor, which allows to shutdown the Executor and to wait for termination.

Create again the Runnable.

			
package de.vogella.concurrency.threadpools;

/**
 * MyRunnable will count the sum of the number from 1 to the parameter
 * countUntil and then write the result to the console.
 * <p>
 * MyRunnable is the task which will be performed
 * 
 * @author Lars Vogel
 * 
 */
public class MyRunnable implements Runnable {
	private final long countUntil;

	MyRunnable(long countUntil) {
		this.countUntil = countUntil;
	}

	@Override
	public void run() {
		long sum = 0;
		for (long i = 1; i < countUntil; i++) {
			sum += i;
		}
		System.out.println(sum);
	}
}

		

Now you run your runnables with the executor framework.

			
package de.vogella.concurrency.threadpools;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
	private static final int NTHREDS = 10;

	public static void main(String[] args) {
		ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
		for (int i = 0; i < 500; i++) {
			Runnable worker = new MyRunnable(10000000L + i);
			executor.execute(worker);
		}
		// This will make the executor accept no new threads
		// and finish all existing threads in the queue
		executor.shutdown();
		// Wait until all threads are finish
		while (!executor.isTerminated()) {

		}
		System.out.println("Finished all threads");
	}
}

		

In case the threads should return some value (result-bearing threads) then you could use java.util.concurrent.Callable.

The executor framework presented in the last chapter works with Runnables. Runnable do not return result.

In case you expect your threads to return a computed result you can use java.util.concurrent.Callable. Callables allow to return values after competition.

Callable uses generic to define the type of object which is returned.

If you submit a callable to an executor the framework returns a java.util.concurrent.Future. This futures can be used to check the status of a callable and to retrieve the result from the callable.

On the executor you can use the method submit to submit a Callable and to get a future. To retrieve the result of the future use the get() method.

			
package de.vogella.concurrency.callables;

import java.util.concurrent.Callable;

public class MyCallable implements Callable<Long> {
	@Override
	public Long call() throws Exception {
		long sum = 0;
		for (long i = 0; i <= 100; i++) {
			sum += i;
		}
		return sum;
	}

}

		

			
package de.vogella.concurrency.callables;

import java.util.ArrayList;
import java.util.List;
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 CallableFutures {
	private static final int NTHREDS = 10;

	public static void main(String[] args) {

		ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
		List<Future<Long>> list = new ArrayList<Future<Long>>();
		for (int i = 0; i < 20000; i++) {
			Callable<Long> worker = new MyCallable();
			Future<Long> submit = executor.submit(worker);
			list.add(submit);
		}
		long sum = 0;
		System.out.println(list.size());
		// Now retrieve the result
		for (Future<Long> future : list) {
			try {
				sum += future.get();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		System.out.println(sum);
	}
}
		

Java 5.0 provides supports for additional atomic operations. This allows to develop algorithm which are non-blocking algorithm, e.g. which do not require synchronization, but are based on low-level atomic hardware primitives such as compare-and-swap (CAS). A compare-and-swap operation check if the variable has a certain value and if it has this value it will perform this operation.

Non-blocking algorithm are usually much faster then blocking algorithms as the synchronization of threads appears on a much finer level (hardware).

For example this created a non-blocking counter which always increases. This example is contained in the project "de.vogella.concurrency.nonblocking.counter".

			
package de.vogella.concurrency.nonblocking.counter;

import java.util.concurrent.atomic.AtomicInteger;

public class Counter {
	private AtomicInteger value = new AtomicInteger(); 
	public int getValue(){
		return value.get();
	}
	public int increment(){
		return value.incrementAndGet();
	}
	
	// Alternative implementation as increment but just make the 
	// implementation explicit
	public int incrementLongVersion(){
		int oldValue = value.get();
		while (value.compareAndSet(value.get(),oldValue+1)){
			 oldValue = value.get();
		}
		return oldValue+1;
	}
	
}

		

And a test.

			
package de.vogella.concurrency.nonblocking.counter;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
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 Test {
		private static final int NTHREDS = 10;

		public static void main(String[] args) {
			final Counter counter = new Counter();
			List<Future<Integer>> list = new ArrayList<Future<Integer>>();

			ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
			for (int i = 0; i < 500; i++) {
				Callable<Integer> worker = new  Callable<Integer>() {
					@Override
					public Integer call() throws Exception {
						int number = counter.increment();
						System.out.println(number );
						return number ;
					}
				};
				Future<Integer> submit= executor.submit(worker);
				list.add(submit);

			}
			
			
			// This will make the executor accept no new threads
			// and finish all existing threads in the queue
			executor.shutdown();
			// Wait until all threads are finish
			while (!executor.isTerminated()) {
			}
			Set<Integer> set = new HashSet<Integer>();
			for (Future<Integer> future : list) {
				try {
					set.add(future.get());
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				}
			}
			if (list.size()!=set.size()){
				throw new RuntimeException("Double-entries!!!"); 
			}

		}


}

		

The interesting part is how incrementAndGet() is implemented. It uses a CAS operation.

			 
public final int incrementAndGet() {
        for (;;) {
            int current = get();
            int next = current + 1;
            if (compareAndSet(current, next))
                return next;
        }
    }
		

The JDK itself makes more and more use of non-blocking algorithms to increase performance for every developer. Developing correct non-blocking algorithm is not a trivial task. For more information on non-blocking algorithm, e.g. examples for a non-blocking Stack and non-block LinkedList, please see http://www.ibm.com/developerworks/java/library/j-jtp04186/index.html

Java 7 introduce a new parallel mechanism for compute intensive tasks, the fork-join framework. The fork-join framework allows you to distribute a certain task on several workers and when wait for the result.

For Java 6.0 you can download the package (jsr166y) from

For testing create the Java project "de.vogella.performance.forkjoin". If you are not using Java 7 you also need to "jsr166y.jar" to the classpath.

Create first a package "algorithm" and then the problem class.

			
package algorithm;

import java.util.Random;

/**
 * 
 * This class defines a long list of integers which defines the problem we will
 * later try to solve
 * 
 */
public class Problem {
	private final int[] list = new int[2000000];

	public Problem() {
		Random generator = new Random(19580427);
		for (int i = 0; i < list.length; i++) {
			list[i] = generator.nextInt(500000);
		}
	}

	public int[] getList() {
		return list;
	}

}

		

Define now the solver class. This class extends RecursiveTask.

			
package algorithm;

import java.util.Arrays;

import jsr166y.forkjoin.RecursiveAction;

public class Solver extends RecursiveAction {
	private int[] list;
	public long result;

	public Solver(int[] array) {
		this.list = array;
	}

	@Override
	protected void compute() {
		if (list.length == 1) {
			result = list[0];
		} else {
			int midpoint = list.length / 2;
			int[] l1 = Arrays.copyOfRange(list, 0, midpoint);
			int[] l2 = Arrays.copyOfRange(list, midpoint, list.length);
			Solver s1 = new Solver(l1);
			Solver s2 = new Solver(l2);
			forkJoin(s1, s2);
			result = s1.result + s2.result;
		}
	}
}

		

Now define a small test class for testing it efficiency.

			
package testing;

import jsr166y.forkjoin.ForkJoinExecutor;
import jsr166y.forkjoin.ForkJoinPool;
import algorithm.Problem;
import algorithm.Solver;

public class Test {

	public static void main(String[] args) {
		Problem test = new Problem();
		// Check the number of available processors
		int nThreads = Runtime.getRuntime().availableProcessors();
		System.out.println(nThreads);
		Solver mfj = new Solver(test.getList());
		ForkJoinExecutor pool = new ForkJoinPool(nThreads);
		pool.invoke(mfj);
		long result = mfj.getResult();
		System.out.println("Done. Result: " + result);
		long sum = 0;
		// Check if the result was ok
		for (int i = 0; i < test.getList().length; i++) {
			sum += test.getList()[i];
		}
		System.out.println("Done. Result: " + sum);
	}
}

		

A concurrent application has the risk of a deadlock. A set of processes are deadlocked if all processes are waiting for an event which another process in the same set has to cause.

For example if thread A waits for a lock on object Z which thread B holds and thread B wait for a look on object Y which is hold be process A then these two processes are looked and cannot continue in their processing.

Thank you for practicing with this tutorial.

Please note that I maintain this website in my private time. If you like the information I'm providing please help me by donating.

For questions and discussion around this article please use the www.vogella.de Google Group. Also if you note an error in this article please post the error and if possible the correction to the Group.

I believe the following is a very good guideline for asking questions in general and also for the Google group How To Ask Questions The Smart Way.