Java Concurrency | Jes2ica.IO

Threads and Runnables

• Before starting a new thread you have to specify the code to be executed by this thread, often called the task. This is done by implementing Runnable - a functional interface defining a single void no-args method run().

  Runnable task = () -> {
      String threadName = Thread.currentThread().getName();
      System.out.println("Hello " + threadName);
  };
  
  task.run();
  
  Thread thread = new Thread(task);
  thread.start();
  
  System.out.println("Done!");

• Due to concurrent execution we cannot predict if the runnable will be invoked before or after printing ‘done’. The order is non-deterministic, thus making concurrent programming a complex task in larger applications.

Executors

• ExecutorService as a higher level replacement for working with threads directly.
• Executors are capable of running asynchronous tasks and typically manage a pool of threads, so we don’t have to create new threads manually.

ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(() -> {
    String threadName = Thread.currentThread().getName();
    System.out.println("Hello " + threadName);
});

Callables and Futures

In addition to Runnable executors support another kind of task named Callable. Callables are functional interfaces just like runnables but instead of being void they return a value.

ExecutorService executor = Executors.newFixedThreadPool(1);
Future<Integer> future = executor.submit(task);

System.out.println("future done? " + future.isDone());

Integer result = future.get();

System.out.println("future done? " + future.isDone());
System.out.print("result: " + result);

Synchronized

When writing such multi-threaded code you have to pay particular attention when accessing shared mutable variables concurrently from multiple threads. Let’s just say we want to increment an integer which is accessible simultaneously from multiple threads.

We define a field count with a method incrementSync() to increase count by one:

int count = 0;

synchronized void incrementSync() {
    count = count + 1;
}

When using incrementSync() concurrently we get the desired result count of 10000. No race conditions occur any longer and the result is stable with every execution of the code:

ExecutorService executor = Executors.newFixedThreadPool(2);

IntStream.range(0, 10000)
    .forEach(i -> executor.submit(this::incrementSync));

stop(executor);

System.out.println(count);  // 10000

Internally Java uses a so called monitor also known as monitor lock or intrinsic lock in order to manage synchronization. This monitor is bound to an object, e.g. when using synchronized methods each method share the same monitor of the corresponding object.

All implicit monitors implement the reentrant characteristics. Reentrant means that locks are bound to the current thread. A thread can safely acquire the same lock multiple times without running into deadlocks (e.g. a synchronized method calls another synchronized method on the same object).

AtomicInteger

An operation is atomic when you can safely perform the operation in parallel on multiple threads without using the synchronized keyword or locks.

Internally, the atomic classes make heavy use of compare-and-swap (CAS), an atomic instruction directly supported by most modern CPUs. Those instructions usually are much faster than synchronizing via locks. So my advice is to prefer atomic classes over locks in case you just have to change a single mutable variable concurrently.

AtomicInteger atomicInt = new AtomicInteger(0);

ExecutorService executor = Executors.newFixedThreadPool(2);

IntStream.range(0, 1000)
    .forEach(i -> executor.submit(atomicInt::incrementAndGet));

stop(executor);

System.out.println(atomicInt.get());    // => 1000

By using AtomicInteger as a replacement for Integer we’re able to increment the number concurrently in a thread-safe manor without synchronizing the access to the variable. The method incrementAndGet() is an atomic operation so we can safely call this method from multiple threads.

AtomicInteger supports various kinds of atomic operations. The method updateAndGet() accepts a lambda expression in order to perform arbitrary arithmetic operations upon the integer:

AtomicInteger atomicInt = new AtomicInteger(0);

ExecutorService executor = Executors.newFixedThreadPool(2);

IntStream.range(0, 1000)
    .forEach(i -> {
        Runnable task = () ->
            atomicInt.updateAndGet(n -> n + 2);
        executor.submit(task);
    });

stop(executor);

System.out.println(atomicInt.get());    // => 2000

ConcurrentMap

The interface ConcurrentMap extends the map interface and defines one of the most useful concurrent collection types. Java 8 introduces functional programming by adding new methods to this interface.

Reference

http://winterbe.com/posts/2015/04/07/java8-concurrency-tutorial-thread-executor-examples/