Types of Concurrency in Java by Oleg Shelajev

Last week on the vJUG, our very own Oleg Shelajev, head of RebelLabs, gave a session called Flavors of Concurrency. It’s an extremely good session which Oleg has given at various conferences and it was time to make it virtual! In this session, Oleg looked at what it means to run in a concurrent fashion, looking at depth into the different models of achieving concurrent code in Java. The session actually came from a blog post which Oleg wrote for RebelLabs back in December 2014 called Flavors of Concurrency in Java: Threads, Executors, ForkJoin and Actors, which is still very relevant today, so make sure you check it out.

Below is the full video of Oleg’s session for your convenience. If you don’t have time to view it all, scroll down and we’ll give you our TL;DR version!

Oleg first talked through the term of concurrency and in particular contrasted it with parallelism which it is often confused with. He showed Joe Armstrong’s great image below.

Parallelism or Concurrency?

The second image shows two coffee machines and two queues, one per coffee machine. The system is running in parallel as it can execute multiple operations simultaneously. On the other hand, the concurrent image shows two queue but only one coffee machine. This means, the queuing can happen in parallel, but the operation of getting a coffee cannot be simultaneous. A concurrent environment is an environment that is capable of potentially performing operations simultaneously, but whether our existing shared resources allow us to do that is another question. Here are some symptoms and possible concerns Concurrency could give us:

• Shared Resources
• Multiple Consumers and Producers
• Out of Order events
• Locks and Deadlocks

What methods of Concurrency do people use today?

Well in the blog post we mentioned earlier in this post, Oleg performed a survey of what models readers used. The results are below and show Executor/Completion Services and bare Threads as the popular choices. Yeah, bare threads! :)

Oleg then looked at each of the concurrency models that are available to us including Threads, Executors, Fork Join, Completable Futures, Actors and Fibers. He takes a simple problem of trying to find an answer to a question by asking lots of search engines on the internet. The result we want to return is the first one that gets returned to us. Oleg looks at each concurrency model to find out how he can solve this problem differently.

Let’s start with the simplest first, Threads.

Threads

On our base operating system, we have processes. These are modelled in the JVM as threads. They have access to the same shared memory, and typically communicate via shared data. To solve the problem, Oleg spawns off many threads and then uses an AtomicReference to store the response from his threads, a construct that allows you to compare the current value before assigning to it. What does this mean to a coder? Well, if we have many threads, which we do, we can compare the AtomicReference value against null and set it, only if the current value is null. This means only the first Thread will write its response, so long as the response isn’t null. Here’s some code:

AtomicReference result = new AtomicReference<>();
 for(String base: engines) {
   String url = base + question;
   new Thread(() -> {
     result.compareAndSet(null, WS.url(url).get());
   }).start();
 }

The problem with threads of course is that they have access to shared data and need to access this in a controlled way. Communication between Threads need to be looked at carefully and this is one of the problems of using a concurrency model at a low level, like Threads. In short, they require manual management.

Executors

The great news about executors is that they’re managed for you! Yay :) You have two flavors with executors, a fire and forget style where you just want to execute a piece of work, as a Runnable but you don’t look for a response and there’s also a completion service style where you do expect a response so you’ll need to implement a callback of sorts. Let’s look at code to make it easier. First here’s the API for Executors:

public interface Executor {
  void execute(Runnable command);
}

public class ExecutorCompletionService implements CompletionService {
…
    public Future submit(Callable task)
    public Future submit(Runnable task, V result)
…
}

The code to create a pool of threads that our executor can use and pass it some work is super simple as follows:

ExecutorCompletionService service = new ExecutorCompletionService(Executors.newFixedThreadPool(4));
 
 for(String base: engines) {
   String url = base + question;
   service.submit(() -> {
     return WS.url(url).get();
   });
 }

To grab the result is super simple again, as shown:

  service.take().get();

The configuration, communication and logic code are now separated. Configuration is easy, and your code is simpler. With the complexity now deeper in your code, you can focus on your business logic.

Fork Join

The focus on the Fork Join framework is to execute recursive tasks. Idle Threads in the Fork Join pool can also help out as needed in a mechanism called work stealing. In Java 8 with the addition of Streams, work has been done to integrate with the new features so we can write code that uses Fork Join implicitly. Check this code out.

Optional result = engines.stream().parallel().map((base) -> {
   String url = base + question;
   return WS.url(url).get();
 }).findAny();

But this is bad code. Why? Well because by making a request out to the internetz is a bit scary. Who knows which dragons might grab out requests and hang on them for a half minute before returning. We don’t want our threads to be blocked for a half minute because we’re taking up resources in our fork join pool which could lead to starvation. These kinds of requests should be fast and ideally non blocking. Again it’s easily configured, well in fact it’s preconfigured, but is easy to get wrong as you’re using it so implicitly it’s easy to overlook global thread pool considerations.

Completable Futures

In Java 8, completable futures care more about how you build and apply code chunks to each thread, rather than the parallelisation itself. Under the covers, completable futures make use of the Fork Join pool, but it adds types which means we can combine functionality that will execute on the result of previous work. In other words we can chain functions onto previous computations. These can be executed synchronously or asynchronously. Here’s some code to show how you’d use a Completable Future.

   CompletableFuture result = CompletableFuture.anyOf(engines.stream().map(
      (base) -> {
        return CompletableFuture.supplyAsync(() -> {
        String URL = base + question;
        return WS.url(url).get();
      });
   }).collect(Collectors.toList()).toArray(new CompletableFuture[0]));

In addition to the code above, should we want to, we could chain computation on to the result by calling one of the following methods on the CompletableFuture type returned:

   thenApply(Function fn)
   thenApplyAsync(Function fn)
   thenApplyAsync(Function fn, Executor executor)

Actors

Actors are lightweight instances that execute pieces of work. They communicate using immutable messages and focus around message passing between async pieces of code. There is no shared mutable state. That’s great, as it removes many of the concurrency problems! One of the most known and commonly used Actor frameworks is Akka. Here are a couple of snippets of the key pieces of Actors, namely, the messages and the results.

static class Message {
 String url;
 Message(String url) {this.url = url;}
}
static class Result {
 String html;
 Result(String html) {this.html = html;}
}

Here we’re using an untyped actor which means we have to call getSender to get the result back. We could also use typed actors which work out the result setting for us. The Actor framework ensures the thread scheduling, message sending and bookkeeping is all done in the background. Actors are not threads. As a result you have have millions of Actors in a system without using up all your resources.

static class Querier extends UntypedActor {
 @Override 
 public void onReceive(Object message) throws Exception {
   if(message instanceof Message) {
     Message work = (Message) message;
     String result = WS.url(work.url).get();
     getSender().tell(new Result(result), getSelf());
   }
   else {
     unhandled(message);
   }
 }
}

Fibers

Want lightweight Threads? You want Fibers. Oleg talks about a project called quasar which provides the ability to suspend a thread between operations. This technique allows you to pause work to pass execution time to another fiber. If you have millions and millions of entities all wanting to make progress at the same time, continuations in fibers allows this making it far more scalable. If you're interested in more details about fibers, check out this blogpost by Fabio Tudone, one of the maintainers of Quasar library. In that post Fabio talks about what Fibers are, how they are implemented and when you want to prefer them to regular threads. Go watch the session, it’s educational, technical and inspiring!

Interview About Concurrency

Finally, I had the pleasure of interviewing Oleg, a job which he himself usually leads. One of our key questions in this interview is “Would you rather be an elephant the size of a dog or a dog the size of an elephant”. I won’t ruin the surprise of Oleg’s answer, so you’ll have to watch to find out.

Additional Resources

Want to learn more about concurrency? Our article, Concurrency Torture: Testing Your Code Within the Java Memory Module is a great read.

View Article