So everything seems to be synchronous. To understand why we need to understand how streams work. The first line creates the source stream, in this case it’s based on an array. The second line adds a mapping operation and thereby creates a mapped stream, a stream of mapped values. As you probably know nothing happens until we subscribe. When we subscribe the following happens:

The subscribe method tells the mapped stream that it’s ready and expects values.

The mapped stream has a parent and forwards the subscription request to it.

The source stream doesn’t have a parent and immediately starts emitting values and continues until it’s completed.

Everything happens on the same thread . That’s why the last statement is executed only after the stream has completed.

Let’s go async

Now let’s replace the source stream with another one: a timer.

source = Flux.interval(Duration.ofSeconds(1))

.take(2); //... // Output:

// Before or after?

// 0. element: 0

// 1. element: 1

We can see that the stream has become asynchronous. Why? Because the source stream doesn’t emit the values immediately. So obviously the source determines whether our stream is synchronous or asynchronous.

But we also can directly influence the threading behavior of our streams. For better illustration let’s again change the source stream:

source = Flux.<Integer>create(emitter -> {

System.out.println(Thread.currentThread().getName());

emitter.next(1);

emitter.complete();

}); // ... // Output:

// main

// 1. element: 1

// Before or after?

We now can see that the source (stream) operates on the main thread. Now let’s add publishOn after the source and execute the code:

source.publishOn(Schedulers.single())

// ... //Output:

// main

// Before or after?

// 1. element: 1

At first everything’s the same as before, but publishOn ensures that every stream operation after it runs on another thread. That’s why the last statement is printed before the values.

Let’s try something different and insert subscribeOn before subscribe:

source.map(i -> i + ". element: " + i)

.subscribeOn(Schedulers.single())

.subscribe(System.out::println); // Output:

// Before or after?

// single-1

// 1. element: 1

As we can see from the output even the subscription process now runs on another thread. We eventually have a truly asynchronous, non-blocking stream.

Conclusion

Reactive streams are not always asynchronous. The threading behavior primarily depends on the source and can be influenced with operators. The nice thing, though, is that the code stays the same. We can use the same programming model and the same pipeline for both synchronous and asynchronous streams.