Async non-blocking code examples for the JVM

Introduction

We are a group of software developers with a monthly meetup in Lucerne (Switzerland). We are interested in programming questions around concurrency, distributed systems, consistency…​

We are learning new ways how to write code that scales up (vertically) and out (horizontally). There are many technologies out there today (libraries, languages, …​), most of them are "pretty new". Some work well for scaling up (RxJava, akka-stream, CompletableFuture, …​), others for scaling out (Vert.x, Hazelcast, akka, …​). Some of them seem to be good for scaling up and out (Vert.x, akka, …​).

A typical core characteristic of these languages and libraries is the "async and non-blocking" working manner.

To get used to "async and non-blocking" and the various technologies we defined a couple of "simple" coding challenges. We try now to implement them and share the solutions.

Have fun in comparing the results and feel free to provide feedback, improvements or contributions via GitHub issues and pull requests.

Sincerly,
the participating members of the Lucerne reactive meetup group

Async non-blocking coding challenges

Rules

• All examples must be implemented without blocking threads

  • except: it’s ok to block the main thread to keep the JVM alive until the example completed

Challenge 1: combining the results of "parallel" calls

                [PriceService1::getPrice]

                          &&

[avg price?] -> [PriceService2::getPrice] -> [println avg price]

                          &&

                [PriceService3::getPrice]

Because the price services in this challenge are "slow", it’s a good idea to call them "in parallel" and combine afterwards the results. It’s important to not block the "avg price?" caller and to do the calculation of the average price asynchronously.

Credits go to https://github.com/christophsuter for his "concurrency notation".

Language	Library	Example
Java	ExecutorService and Futures (JDK 5)
Java	Google Guava ListenableFuture	./src/main/java/challenge1/guava/ListenableFutureExample.java
Java	CompletableFuture (JDK 8)	./src/main/java/challenge1/java8/CompletableFutureExample.java ./src/main/java/challenge1/java8/GmoCompletableFutureDemo.java
Java	Parallel Streams (JDK 8)	./src/main/java/challenge1/java8/Java8StreamsExample.java
Java	RxJava 1 Observable	./src/main/java/challenge1/rxjava1/RxJavaObservableExample.java
Java	RxJava 2 Flowable	./src/main/java/challenge1/rxjava2/RxJava2Example.java
Java	reactor-core Flux
Java	JDK 9 API Flow / SubmissionPublisher (jsr166-4jdk8)
Scala	Scala 1.12 API scala-futures	./src/main/scala/challenge1/scala_futures/ScalaFutureExample.scala ./src/main/scala/challenge1/scala_futures/ScalaFutureMoreCallsExample.scala
Scala	akka-stream	./src/main/scala/challenge1/akka_stream/AkkaStreamExample.scala ./src/main/scala/challenge1/akka_stream/AkkaStreamFlowExample.scala ./src/main/scala/challenge1/akka_stream/AkkaStreamGraphExample.scala
Java	akka-stream
Scala	akka (actors)	./src/main/scala/challenge1/akka_actor/AkkaActorExample.scala
Java	akka (actors)
Java	Vert.x
Java	cyclops-react (Intro)
Kotlin	Coroutines	./src/main/kotlin/challenge1/coroutines/KotlinCoroutinesExample.kt
…​	…​ yours?

Challenge 2: fallback in case of timeout

                      [PriceService::getPrice ... ... ... ... ... ... ]

[current price?] ->             ||      -> [println price or fallback value]

                             [timeout]

Sometimes things don’t work as expected. In this challenge here the price service is extra slow and "takes forever" to deliver the price. Because we can’t fix the price service we use instead a fallback mechanism to keep our code "responsive": If the price service doesn’t answer within 2 seconds, we provide a fallback value of 42.

Don’t forget to "cancel" the price service in case of timeouts.

Language	Library	Example
Java	ExecutorService and Futures (JDK 5)
Java	Google Guava ListenableFuture
Java	CompletableFuture (JDK 8)	./src/main/java/challenge2/java8/GmoCompletableFutureDemo.java ./src/main/java/challenge2/java8/GmoCompletableFutureDemo2.java
Java	RxJava 1 Observable
Java	RxJava 2 Flowable
Java	reactor-core Flux
Java	JDK 9 API Flow / SubmissionPublisher (jsr166-4jdk8)	./src/main/java/challenge2/java9/GmoCompletableFutureDemoJava9.java
Scala	scala-futures	./src/main/scala/challenge2/scala_futures/ScalaFutureExample.scala ./src/main/scala/challenge2/scala_futures/ScalaFutureMoreCallsExample.scala
Scala	akka-stream	./src/main/scala/challenge2/akka_stream/AkkaStreamExample.scala ./src/main/scala/challenge2/akka_stream/AkkaMultiStreamExample.scala
Java	akka-stream
Scala	akka (actors)	./src/main/scala/challenge2/akka_actor/AkkaActorExample.scala
Java	akka (actors)
Java	Vert.x
Java	cyclops-react (Intro)
Java	Apache Camel	./src/main/java/challenge2/camel/Challenge2.java ./src/main/java/challenge2/camel/Challenge2RouteBuilder.java
…​	…​
…​	…​ yours?

Challenge 3: stream of temperature values

[T1] ... [T2] ... [T3] ... [T4] ... [T5] ... [T6] ... [T7] ... [T8] ... [T9] ...

                                   \                                         \
                                    -> [println min max]                      -> [println min max]

We have a source of data. It pushes every 2 seconds a new value Tx (the current temperature) to its observers. We have one observer which prints out every 10 seconds the min and the max temperature.

Language	Library	Example
Java	ExecutorService and Futures (JDK 5)	./src/main/java/challenge3/java5/Challenge3Java5Example.java
Java	Google Guava ListenableFuture
Java	CompletableFuture (JDK 8)
Java	RxJava 1 Observable
Java	RxJava 2 Flowable / Observable	./src/main/java/challenge3/rxjava2/Challenge3RxJava2Example.java ./src/main/java/challenge3/rxjava2/Challenge3RxJava2ExampleWithThrotteling.java
Java	reactor-core Flux
Java	JDK 9 API Flow / SubmissionPublisher (jsr166-4jdk8)
Scala	akka-stream	./src/main/scala/challenge3/akka_stream/AkkaStreamExample.scala
Java	akka-stream
Scala	akka (actors)	./src/main/scala/challenge3/akka_actor/AkkaActorExample.scala
Java	akka (actors)
Java	Vert.x
Java	cyclops-react (Intro)
Java	Apache Camel	./src/main/java/challenge3/camel/Challenge3.java ./src/main/java/challenge3/camel/Challenge3RouteBuilder.java
…​	…​
…​	…​ yours?

Challenge 4: request collapsing

        [current price?] ->                    -> [println price]
                            \                /

[current price?] -> [PriceService1::getPrice]  -> [println price]

Because calls to our price service are "expensive", we’d like to "collapse" concurrent requests. This is a little bit like caching, but instead of just sharing the results we share the ongoing calculation of a result.

Language	Library	Example
Java	ExecutorService and Futures (JDK 5)
Java	Google Guava ListenableFuture
Java	CompletableFuture (JDK 8)	./src/main/java/challenge4/java8/GmoCompletableFutureDemo.java
Java	RxJava 1 Single / Observable	./src/main/java/challenge4/rxjava1/RequestCollapsingWithShareOperatorAndAtomicReference.java
Java	RxJava 2 Flowable	./src/main/java/challenge4/rxjava2/CollapsingWIthRxJava2Example.java
Java	reactor-core Flux
Java	JDK 9 API Flow / SubmissionPublisher (jsr166-4jdk8)
Scala	akka-stream
Java	akka-stream
Scala	akka (actors)	./src/main/scala/challenge4/akka_actor/AkkaActorExample.scala
Java	akka (actors)	./src/main/java/challenge4/akka_actor/CollapsingWithAkka.java
Java	Vert.x
Java	cyclops-react (Intro)
Java	Apache Camel	./src/main/java/challenge4/camel/Challenge4.java ./src/main/java/challenge4/camel/Challenge4RouteBuilder.java
…​	…​
…​	…​ yours?