Reactor 3 Reference Guide

The Reactor reference guide is available as HTML documents. The latest copy is available at https://projectreactor.io/docs/core/release/reference/index.html

Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.

The reference guide is written in Asciidoc, and you can find its sources at https://github.com/reactor/reactor-core/tree/main/docs/asciidoc.

If you have an improvement or a suggestion, we will be happy to get a pull request from you!

We recommend that you check out a local copy of the repository so that you can generate the documentation by running the asciidoctor gradle task and checking the rendering. Some of the sections rely on included files, so GitHub rendering is not always complete.

You can reach out for help in several ways with Reactor:

Suggest Edit to "About the Documentation"

Reactor is a fully non-blocking reactive programming foundation for the JVM, with efficient demand management (in the form of managing “backpressure”). It integrates directly with the Java 8 functional APIs, notably CompletableFuture, Stream, and Duration. It offers composable asynchronous sequence APIs — Flux (for [N] elements) and Mono (for [0|1] elements) — and extensively implements the Reactive Streams specification.

Reactor also supports non-blocking inter-process communication with the reactor-netty project. Suited for Microservices Architecture, Reactor Netty offers backpressure-ready network engines for HTTP (including Websockets), TCP, and UDP. Reactive encoding and decoding are fully supported.

Reactor Core runs on Java 8 and above.

It has a transitive dependency on org.reactivestreams:reactive-streams:1.0.3.

Reactor 3 uses a BOM (Bill of Materials) model (since reactor-core 3.0.4, with the Aluminium release train). This curated list groups artifacts that are meant to work well together, providing the relevant versions despite potentially divergent versioning schemes in these artifacts.

Note the versioning scheme has changed between 3.3.x and 3.4.x (Dysprosium and Europium).

Artifacts follow a versioning scheme of MAJOR.MINOR.PATCH-QUALIFIER while the BOM is versioned using a CalVer inspired scheme of YYYY.MINOR.PATCH-QUALIFIER, where:

The first release cycle to follow that convention is thus 2020.0.x, codename Europium. The scheme uses the following qualifiers (note the use of dash separator), in order:

Each release cycle is also given a codename, in continuity with the previous codename-based scheme, which can be used to reference it more informally (like in discussions, blog posts, etc…​). The codenames represent what would traditionally be the MAJOR.MINOR number. They (mostly) come from the Periodic Table of Elements, in increasing alphabetical order.

As mentioned earlier, the easiest way to use Reactor in your core is to use the BOM and add the relevant dependencies to your project. Note that, when you add such a dependency, you must omit the version so that the version gets picked up from the BOM.

However, if you want to force the use of a specific artifact’s version, you can specify it when adding your dependency, as you usually would. You can also forgo the BOM entirely and specify dependencies by their artifact versions.

The entries below are mirroring https://github.com/reactor/.github/blob/main/SUPPORT.adoc

Modern applications can reach huge numbers of concurrent users, and, even though the capabilities of modern hardware have continued to improve, performance of modern software is still a key concern.

There are, broadly, two ways one can improve a program’s performance:

Usually, Java developers write programs by using blocking code. This practice is fine until there is a performance bottleneck. Then it is time to introduce additional threads, running similar blocking code. But this scaling in resource utilization can quickly introduce contention and concurrency problems.

Worse still, blocking wastes resources. If you look closely, as soon as a program involves some latency (notably I/O, such as a database request or a network call), resources are wasted because threads (possibly many threads) now sit idle, waiting for data.

So the parallelization approach is not a silver bullet. It is necessary to access the full power of the hardware, but it is also complex to reason about and susceptible to resource wasting.

The second approach mentioned earlier, seeking more efficiency, can be a solution to the resource wasting problem. By writing asynchronous, non-blocking code, you let the execution switch to another active task that uses the same underlying resources and later comes back to the current process when the asynchronous processing has finished.

But how can you produce asynchronous code on the JVM? Java offers two models of asynchronous programming:

Are these techniques good enough? Not for every use case, and both approaches have limitations.

Callbacks are hard to compose together, quickly leading to code that is difficult to read and maintain (known as “Callback Hell”).

Consider an example: showing the top five favorites from a user on the UI or suggestions if she does not have a favorite. This goes through three services (one gives favorite IDs, the second fetches favorite details, and the third offers suggestions with details), as follows:

That is a lot of code, and it is a bit hard to follow and has repetitive parts. Consider its equivalent in Reactor:

What if you want to ensure the favorite IDs are retrieved in less than 800ms or, if it takes longer, get them from a cache? In the callback-based code, that is a complicated task. In Reactor it becomes as easy as adding a timeout operator in the chain, as follows:

Future objects are a bit better than callbacks, but they still do not do well at composition, despite the improvements brought in Java 8 by CompletableFuture. Orchestrating multiple Future objects together is doable but not easy. Also, Future has other problems:

Consider another example: We get a list of IDs from which we want to fetch a name and a statistic and combine these pair-wise, all of it asynchronously. The following example does so with a list of type CompletableFuture:

Since Reactor has more combination operators out of the box, this process can be simplified, as follows:

The perils of using callbacks and Future objects are similar and are what reactive programming addresses with the Publisher-Subscriber pair.

Reactive libraries, such as Reactor, aim to address these drawbacks of “classic” asynchronous approaches on the JVM while also focusing on a few additional aspects:

The following image shows how a Flux transforms items:

A Flux<T> is a standard Publisher<T> that represents an asynchronous sequence of 0 to N emitted items, optionally terminated by either a completion signal or an error. As in the Reactive Streams spec, these three types of signal translate to calls to a downstream Subscriber’s onNext, onComplete, and onError methods.

With this large scope of possible signals, Flux is the general-purpose reactive type. Note that all events, even terminating ones, are optional: no onNext event but an onComplete event represents an empty finite sequence, but remove the onComplete and you have an infinite empty sequence (not particularly useful, except for tests around cancellation). Similarly, infinite sequences are not necessarily empty. For example, Flux.interval(Duration) produces a Flux<Long> that is infinite and emits regular ticks from a clock.

The following image shows how a Mono transforms an item:

A Mono<T> is a specialized Publisher<T> that emits at most one item via the onNext signal then terminates with an onComplete signal (successful Mono, with or without value), or only emits a single onError signal (failed Mono).

Most Mono implementations are expected to immediately call onComplete on their Subscriber after having called onNext. Mono.never() is an outlier: it doesn’t emit any signal, which is not technically forbidden although not terribly useful outside of tests. On the other hand, a combination of onNext and onError is explicitly forbidden.

Mono offers only a subset of the operators that are available for a Flux, and some operators (notably those that combine the Mono with another Publisher) switch to a Flux. For example, Mono#concatWith(Publisher) returns a Flux while Mono#then(Mono) returns another Mono.

Note that you can use a Mono to represent no-value asynchronous processes that only have the concept of completion (similar to a Runnable). To create one, you can use an empty Mono<Void>.

The easiest way to get started with Flux and Mono is to use one of the numerous factory methods found in their respective classes.

For instance, to create a sequence of String, you can either enumerate them or put them in a collection and create the Flux from it, as follows:

Other examples of factory methods include the following:

When it comes to subscribing, Flux and Mono make use of Java 8 lambdas. You have a wide choice of .subscribe() variants that take lambdas for different combinations of callbacks, as shown in the following method signatures:

In this section, we introduce the creation of a Flux or a Mono by programmatically defining its associated events (onNext, onError, and onComplete). All these methods share the fact that they expose an API to trigger the events that we call a sink. There are actually a few sink variants, which we’ll get to shortly.

Reactor, like RxJava, can be considered to be concurrency-agnostic. That is, it does not enforce a concurrency model. Rather, it leaves you, the developer, in command. However, that does not prevent the library from helping you with concurrency.

Obtaining a Flux or a Mono does not necessarily mean that it runs in a dedicated Thread. Instead, most operators continue working in the Thread on which the previous operator executed. Unless specified, the topmost operator (the source) itself runs on the Thread in which the subscribe() call was made. The following example runs a Mono in a new thread:

The preceding code produces the following output:

In Reactor, the execution model and where the execution happens is determined by the Scheduler that is used. A Scheduler has scheduling responsibilities similar to an ExecutorService, but having a dedicated abstraction lets it do more, notably acting as a clock and enabling a wider range of implementations (virtual time for tests, trampolining or immediate scheduling, and so on).

The Schedulers class has static methods that give access to the following execution contexts:

Additionally, you can create a Scheduler out of any pre-existing ExecutorService by using Schedulers.fromExecutorService(ExecutorService). (You can also create one from an Executor, although doing so is discouraged.)

You can also create new instances of the various scheduler types by using the newXXX methods. For example, Schedulers.newParallel(yourScheduleName) creates a new parallel scheduler named yourScheduleName.

Some operators use a specific scheduler from Schedulers by default (and usually give you the option of providing a different one). For instance, calling the Flux.interval(Duration.ofMillis(300)) factory method produces a Flux<Long> that ticks every 300ms. By default, this is enabled by Schedulers.parallel(). The following line changes the Scheduler to a new instance similar to Schedulers.single():

Reactor offers two means of switching the execution context (or Scheduler) in a reactive chain: publishOn and subscribeOn. Both take a Scheduler and let you switch the execution context to that scheduler. But the placement of publishOn in the chain matters, while the placement of subscribeOn does not. To understand that difference, you first have to remember that nothing happens until you subscribe.

In Reactor, when you chain operators, you can wrap as many Flux and Mono implementations inside one another as you need. Once you subscribe, a chain of Subscriber objects is created, backward (up the chain) to the first publisher. This is effectively hidden from you. All you can see is the outer layer of Flux (or Mono) and Subscription, but these intermediate operator-specific subscribers are where the real work happens.

With that knowledge, we can have a closer look at the publishOn and subscribeOn operators:

In Reactive Streams, errors are terminal events. As soon as an error occurs, it stops the sequence and gets propagated down the chain of operators to the last step, the Subscriber you defined and its onError method.

Such errors should still be dealt with at the application level. For instance, you might display an error notification in a UI or send a meaningful error payload in a REST endpoint. For this reason, the subscriber’s onError method should always be defined.

Reactor also offers alternative means of dealing with errors in the middle of the chain, as error-handling operators. The following example shows how to do so:

Now we can consider each means of error handling one-by-one. When relevant, we make a parallel with imperative programming’s try patterns.

In Reactor a sink is a class that allows safe manual triggering of signals in a standalone fashion, creating a Publisher-like structure capable of dealing with multiple Subscriber (with the exception of unicast() flavors).

Before 3.5.0, there was also a set of Processor implementations which has been phased out.

Reactor supports Kotlin 1.1+ and requires kotlin-stdlib (or one of its kotlin-stdlib-jdk7 or kotlin-stdlib-jdk8 variants).

Thanks to its great Java interoperability and to Kotlin extensions, Reactor Kotlin APIs leverage regular Java APIs and are additionally enhanced by a few Kotlin-specific APIs that are available out of the box within Reactor artifacts.

For example, Kotlin reified type parameters provide a workaround for JVM generics type erasure, and Reactor provides some extensions to take advantage of this feature.

The following table compares Reactor with Java against Reactor with Kotlin and extensions:

The Reactor KDoc API lists and documents all the available Kotlin extensions.

One of Kotlin’s key features is null safety, which cleanly deals with null values at compile time rather than bumping into the famous NullPointerException at runtime. This makes applications safer through nullability declarations and expressive “value or no value” semantics without paying the cost of wrappers such as Optional. (Kotlin allows using functional constructs with nullable values. See this comprehensive guide to Kotlin null-safety.)

Although Java does not let one express null safety in its type-system, Reactor now provides null safety of the whole Reactor API through tooling-friendly annotations declared in the reactor.util.annotation package. By default, types from Java APIs used in Kotlin are recognized as platform types for which null-checks are relaxed. Kotlin support for JSR 305 annotations and Reactor nullability annotations provide null-safety for the whole Reactor API to Kotlin developers, with the advantage of dealing with null-related issues at compile time.

You can configure the JSR 305 checks by adding the -Xjsr305 compiler flag with the following options: -Xjsr305={strict|warn|ignore}.

For kotlin versions 1.1.50+, the default behavior is the same as -Xjsr305=warn. The strict value is required to have the Reactor API full null-safety taken into account but should be considered experimental, since the Reactor API nullability declaration could evolve even between minor releases, as more checks may be added in the future).

Suggest Edit to "Kotlin support"

The most common case for testing a Reactor sequence is to have a Flux or a Mono defined in your code (for example, it might be returned by a method) and to want to test how it behaves when subscribed to.

This situation translates well to defining a “test scenario,” where you define your expectations in terms of events, step-by-step. You can ask and answer questions such as the following:

You can express all of that through the StepVerifier API.

For instance, you could have the following utility method in your codebase that decorates a Flux:

In order to test it, you want to verify the following scenario:

In the StepVerifier API, this translates to the following test:

The API is a builder. You start by creating a StepVerifier and passing the sequence to be tested. This offers a choice of methods that let you:

For terminal events, the corresponding expectation methods (expectComplete() and expectError() and all their variants) switch to an API where you cannot express expectations anymore. In that last step, all you can do is perform some additional configuration on the StepVerifier and then trigger the verification, often with verify() or one of its variants.

What happens at this point is that the StepVerifier subscribes to the tested Flux or Mono and plays the sequence, comparing each new signal with the next step in the scenario. As long as these match, the test is considered a success. As soon as there is a discrepancy, an AssertionError is thrown.

Note that, if one of the lambda-based expectations throws an AssertionError, it is reported as is, failing the test. This is useful for custom assertions.

You can use StepVerifier with time-based operators to avoid long run times for corresponding tests. You can do so through the StepVerifier.withVirtualTime builder.

It looks like the following example:

This virtual time feature plugs in a custom Scheduler in Reactor’s Schedulers factory. Since these timed operators usually use the default Schedulers.parallel() scheduler, replacing it with a VirtualTimeScheduler does the trick. However, an important prerequisite is that the operator be instantiated after the virtual time scheduler has been activated.

To increase the chances that this happens correctly, the StepVerifier does not take a simple Flux as input. withVirtualTime takes a Supplier, which guides you into lazily creating the instance of the tested flux after having done the scheduler set up.

There are two expectation methods that deal with time, and they are both valid with or without virtual time:

Both methods pause the thread for the given duration in classic mode and advance the virtual clock instead in virtual mode.

In order to quickly evaluate the behavior of our Mono.delay above, we can finish writing our code as follows:

We could have used thenAwait(Duration.ofDays(1)) above, but expectNoEvent has the benefit of guaranteeing that nothing happened earlier than it should have.

Note that verify() returns a Duration value. This is the real-time duration of the entire test.

After having described the final expectation of your scenario, you can switch to a complementary assertion API instead of triggering verify(). To do so, use verifyThenAssertThat() instead.

verifyThenAssertThat() returns a StepVerifier.Assertions object, which you can use to assert a few elements of state once the whole scenario has played out successfully (because it also calls verify()). Typical (albeit advanced) usage is to capture elements that have been dropped by some operator and assert them (see the section on Hooks).

For more information about the Context, see Adding a Context to a Reactive Sequence.

StepVerifier comes with a couple of expectations around the propagation of a Context:

Additionally, you can associate a test-specific initial Context to a StepVerifier by using StepVerifierOptions to create the verifier.

These features are demonstrated in the following snippet:

For more advanced test cases, it might be useful to have complete mastery over the source of data, to trigger finely chosen signals that closely match the particular situation you want to test.

Another situation is when you have implemented your own operator and you want to verify how it behaves with regards to the Reactive Streams specification, especially if its source is not well behaved.

For both cases, reactor-test offers the TestPublisher class. This is a Publisher<T> that lets you programmatically trigger various signals:

You can get a well behaved TestPublisher through the create factory method. Also, you can create a misbehaving TestPublisher by using the createNonCompliant factory method. The latter takes a value or multiple values from the TestPublisher.Violation enum. The values define which parts of the specification the publisher can overlook. These enum values include:

Finally, the TestPublisher keeps track of internal state after subscription, which can be asserted through its various assert* methods.

You can use it as a Flux or Mono by using the conversion methods, flux() and mono().

When building complex chains of operators, you could come across cases where there are several possible execution paths, materialized by distinct sub-sequences.

Most of the time, these sub-sequences produce a specific-enough onNext signal that you can assert that it was executed by looking at the end result.

For instance, consider the following method, which builds a chain of operators from a source and uses a switchIfEmpty to fall back to a particular alternative if the source is empty:

You can test which logical branch of the switchIfEmpty was used, as follows:

However, think about an example where the method produces a Mono<Void> instead. It waits for the source to complete, performs an additional task, and completes. If the source is empty, a fallback Runnable-like task must be performed instead. The following example shows such a case:

To verify that your processOrFallback method does indeed go through the doWhenEmpty path, you need to write a bit of boilerplate. Namely you need a Mono<Void> that:

Before version 3.1, you would need to manually maintain one AtomicBoolean per state you wanted to assert and attach a corresponding doOn* callback to the publisher you wanted to evaluate. This could be a lot of boilerplate when having to apply this pattern regularly. Fortunately, 3.1.0 introduced an alternative with PublisherProbe. The following example shows how to use it:

You can also use the probe in place of a Flux<T> by calling .flux() instead of .mono(). For cases where you need to probe an execution path but also need the probe to emit data, you can wrap any Publisher<T> by using PublisherProbe.of(Publisher).

Suggest Edit to "Testing"

When you shift to asynchronous code, things can get much more complicated.

Consider the following stack trace:

There is a lot going on there. We get an IndexOutOfBoundsException, which tells us that a source emitted more than one item.

We can probably quickly come to assume that this source is a Flux or a Mono, as confirmed by the next line, which mentions MonoSingle. So it appears to be some sort of complaint from a single operator.

Referring to the Javadoc for the Mono#single operator, we see that single has a contract: The source must emit exactly one element. It appears we had a source that emitted more than one and thus violated that contract.

Can we dig deeper and identify that source? The following rows are not very helpful. They take us through the internals of what seems to be a reactive chain, through multiple calls to subscribe and request.

By skimming over these rows, we can at least start to form a picture of the kind of chain that went wrong: It seems to involve a MonoSingle, a FluxFlatMap, and a FluxRange (each gets several rows in the trace, but overall these three classes are involved). So a range().flatMap().single() chain maybe?

But what if we use that pattern a lot in our application? This still does not tell us much, and simply searching for single is not going to find the problem. Then the last line refers to some of our code. Finally, we are getting close.

Hold on, though. When we go to the source file, all we see is that a pre-existing Flux is subscribed to, as follows:

All of this happened at subscription time, but the Flux itself was not declared there. Worse, when we go to where the variable is declared, we see the following:

The variable is not instantiated where it is declared. We must assume a worst-case scenario where we find out that there could be a few different code paths that set it in the application. We remain unsure of which one caused the issue.

What we want to find out more easily is where the operator was added into the chain - that is, where the Flux was declared. We usually refer to that as the “assembly” of the Flux.

Even though the stacktrace was still able to convey some information for someone with a bit of experience, we can see that it is not ideal by itself in more advanced cases.

Fortunately, Reactor comes with assembly-time instrumentation that is designed for debugging.

This is done by activating a global debug mode via the Hooks.onOperatorDebug() method at application start (or at least before the incriminated Flux or Mono can be instantiated), as follows:

This starts instrumenting the calls to Reactor operator methods (where they are assembled into the chain) by wrapping the construction of the operator and capturing a stack trace there. Since this is done when the operator chain is declared, the hook should be activated before that, so the safest way is to activate it right at the start of your application.

Later on, if an exception occurs, the failing operator is able to refer to that capture and to rework the stack trace, appending additional information.

In the next section, we see how the stack trace differs and how to interpret that new information.

When we reuse our initial example but activate the operatorStacktrace debug feature, several things happen:

The full stack trace, once printed, is as follows:

The captured stack trace is appended to the original error as a suppressed OnAssemblyException. There are three parts to it, but the first section is the most interesting. It shows the path of construction for the operator that triggered the exception. Here, it shows that the single that caused our issue was actually created in the scatterAndGather method.

Now that we are armed with enough information to find the culprit, we can have a meaningful look at that scatterAndGather method:

Now we can see what the root cause of the error was a flatMap that performs several HTTP calls to a few URLs but that is chained with single, which is too restrictive. After a short git blame and a quick discussion with the author of that line, we find out he meant to use the less restrictive take(1) instead.

We have solved our problem.

Now consider the following section in the stack trace:

That second part of the traceback was not necessarily interesting in this particular example, because the error was actually happening in the last operator in the chain (the one closest to subscribe). Considering another example might make it more clear:

Now imagine that, inside findAllUserByName, there is a map that fails. Here, we would see the following in the second part of the traceback:

This corresponds to the section of the chain(s) of operators that gets notified of the error:

In some cases where the same exception is propagated through multiple chains, the "root" marker *_ allows us to better separate such chains. If a site is seen several time, there will be an (observed x times) after the call site information.

For instance, let us consider the following snippet:

In the code above, error propagates to the when, going through two separate chains chain1 and chain2. It would lead to a traceback containing the following:

We see that:

We deal with a form of instrumentation here, and creating a stack trace is costly. That is why this debugging feature should only be activated in a controlled manner, as a last resort.

Project Reactor comes with a separate Java Agent that instruments your code and adds debugging info without paying the cost of capturing the stacktrace on every operator call. The behaviour is very similar to Activating Debug Mode - aka tracebacks, but without the runtime performance overhead.

To use it in your app, you must add it as a dependency.

The following example shows how to add reactor-tools as a dependency in Maven:

The following example shows how to add reactor-tools as a dependency in Gradle:

It also needs to be explicitly initialized with:

You may also re-process existing classes with processExistingClasses() if you cannot run the init eagerly. For example, in JUnit5 tests from a TestExecutionListener or even in the class static initializer block:

In addition to stack trace debugging and analysis, another powerful tool to have in your toolkit is the ability to trace and log events in an asynchronous sequence.

The log() operator can do just that. Chained inside a sequence, it peeks at every event of the Flux or Mono upstream of it (including onNext, onError, and onComplete as well as subscriptions, cancellations, and requests).

For instance, suppose we have Logback activated and configured and a chain like range(1,10).take(3). By placing a log() before the take, we can get some insight into how it works and what kind of events it propagates upstream to the range, as the following example shows:

This prints out the following (through the logger’s console appender):

The second (2) and last lines (4) are the most interesting. We can see the take in action there. It leverages backpressure in order to ask the source for exactly the expected amount of elements. After having received enough elements, it tells the source no more items will be needed by calling cancel(). Note that if downstream had itself used backpressure, eg. by requesting only 1 element, the take operator would have honored that (it caps the request when propagating it from downstream to upstream).

Suggest Edit to "Debugging Reactor"

Every async operation in Reactor is done via the Scheduler abstraction described in Threading and Schedulers. This is why it is important to monitor your schedulers, watch out for key metrics that start to look suspicious and react accordingly.

The reactor-core-micrometer module offers a "timed" Scheduler wrapper that perform measurements around tasks submitted through it, which can be used as follows:

See Micrometer.timedScheduler() for produced meters and associated default tags.

Sometimes it is useful to be able to record metrics at some stage in your reactive pipeline.

One way to do it would be to manually push the values to your metrics backend of choice from a custom SignalListener provided to the tap operator.

An out-of-the-box implementation is actually provided by the reactor-core-micrometer module, via Micrometer#metrics APIs. Consider the following pipeline:

To enable the metrics for this source Flux (returned from listenToEvents()), we need to turn on the metrics collection:

The detail of the exposed metrics is available in Micrometer.metrics().

From a clean-code perspective, code reuse is generally a good thing. Reactor offers a few patterns that can help you reuse and mutualize code, notably for operators or combinations of operators that you might want to apply regularly in your codebase. If you think of a chain of operators as a recipe, you can create a “cookbook” of operator recipes.

So far, we have considered that all Flux (and Mono) are the same: They all represent an asynchronous sequence of data, and nothing happens before you subscribe.

Really, though, there are two broad families of publishers: hot and cold.

The earlier description applies to the cold family of publishers. They generate data anew for each subscription. If no subscription is created, data never gets generated.

Think of an HTTP request: Each new subscriber triggers an HTTP call, but no call is made if no one is interested in the result.

Hot publishers, on the other hand, do not depend on any number of subscribers. They might start publishing data right away and would continue doing so whenever a new Subscriber comes in (in which case, the subscriber would see only new elements emitted after it subscribed). For hot publishers, something does indeed happen before you subscribe.

One example of the few hot operators in Reactor is just: It directly captures the value at assembly time and replays it to anybody subscribing to it later. To re-use the HTTP call analogy, if the captured data is the result of an HTTP call, then only one network call is made, when instantiating just.

To transform just into a cold publisher, you can use defer. It defers the HTTP request in our example to subscription time (and would result in a separate network call for each new subscription).

On the opposite, share() and replay(…​) can be used to turn a cold publisher into a hot one (at least once a first subscription has happened). Both of these also have Sinks.Many equivalents in the Sinks class, which allow programmatically feeding the sequence.

Consider two examples, one that demonstrates a cold Flux and the other that makes use of the Sinks to simulate a hot Flux. The following code shows the first example:

This first example produces the following output:

The following image shows the replay behavior:

Both subscribers catch all four colors, because each subscriber causes the process defined by the operators on the Flux to run.

Compare the first example to the second example, shown in the following code:

The second example produces the following output:

The following image shows how a subscription is broadcast:

Subscriber 1 catches all four colors. Subscriber 2, having been created after the first two colors were produced, catches only the last two colors. This difference accounts for the doubling of ORANGE and PURPLE in the output. The process described by the operators on this Flux runs regardless of when subscriptions have been attached.

Sometimes, you may want to not defer only some processing to the subscription time of one subscriber, but you might actually want for several of them to rendezvous and then trigger the subscription and data generation.

This is what ConnectableFlux is made for. Two main patterns are covered in the Flux API that return a ConnectableFlux: publish and replay.

A ConnectableFlux offers additional methods to manage subscriptions downstream versus subscriptions to the original source. These additional methods include the following:

Consider the following example:

The preceding code produces the following output:

The following code uses autoConnect:

The preceding code produces the following output:

When you have lots of elements and you want to separate them into batches, you have three broad solutions in Reactor: grouping, windowing, and buffering. These three are conceptually close, because they redistribute a Flux<T> into an aggregate. Grouping and windowing create a Flux<Flux<T>>, while buffering aggregates into a Collection<T>.

With multi-core architectures being a commodity nowadays, being able to easily parallelize work is important. Reactor helps with that by providing a special type, ParallelFlux, that exposes operators that are optimized for parallelized work.

To obtain a ParallelFlux, you can use the parallel() operator on any Flux. By itself, this method does not parallelize the work. Rather, it divides the workload into “rails” (by default, as many rails as there are CPU cores).

In order to tell the resulting ParallelFlux where to run each rail (and, by extension, to run rails in parallel) you have to use runOn(Scheduler). Note that there is a recommended dedicated Scheduler for parallel work: Schedulers.parallel().

Compare the next two examples:

The first example produces the following output:

The second correctly parallelizes on two threads, as shown in the following output:

If, once you process your sequence in parallel, you want to revert back to a “normal” Flux and apply the rest of the operator chain in a sequential manner, you can use the sequential() method on ParallelFlux.

Note that sequential() is implicitly applied if you subscribe to the ParallelFlux with a Subscriber but not when using the lambda-based variants of subscribe.

Note also that subscribe(Subscriber<T>) merges all the rails, while subscribe(Consumer<T>) runs all the rails. If the subscribe() method has a lambda, each lambda is executed as many times as there are rails.

You can also access individual rails or “groups” as a Flux<GroupedFlux<T>> through the groups() method and apply additional operators to them through the composeGroup() method.

As we described in the Threading and Schedulers section, Reactor Core comes with several Scheduler implementations. While you can always create new instances through the new* factory methods, each Scheduler flavor also has a default singleton instance that is accessible through the direct factory method (such as Schedulers.boundedElastic() versus Schedulers.newBoundedElastic(…​)).

These default instances are the ones used by operators that need a Scheduler to work when you do not explicitly specify one. For example, Flux#delayElements(Duration) uses the Schedulers.parallel() instance.

In some cases, however, you might need to change these default instances with something else in a cross-cutting way, without having to make sure every single operator you call has your specific Scheduler as a parameter. An example is measuring the time every single scheduled task takes by wrapping the real schedulers, for instrumentation purposes. In other words, you might want to change the default Schedulers.

Changing the default schedulers is possible through the Schedulers.Factory class. By default, a Factory creates all the standard Scheduler through similarly named methods. You can override each of these with your custom implementation.

Additionally, the factory exposes one additional customization method: decorateExecutorService. It is invoked during the creation of every Reactor Core Scheduler that is backed by a ScheduledExecutorService (even non-default instances, such as those created by calls to Schedulers.newParallel()).

This lets you tune the ScheduledExecutorService to be used: The default one is exposed as a Supplier and, depending on the type of Scheduler being configured, you can choose to entirely bypass that supplier and return your own instance or you can get() the default instance and wrap it.

Finally, there is a last customizable hook in Schedulers: onHandleError. This hook is invoked whenever a Runnable task submitted to a Scheduler throws an Exception (note that if there is an UncaughtExceptionHandler set for the Thread that ran the task, both the handler and the hook are invoked).

Reactor has another category of configurable callbacks that are invoked by Reactor operators in various situations. They are all set in the Hooks class, and they fall into three categories:

One of the big technical challenges encountered when switching from an imperative programming perspective to a reactive programming mindset lies in how you deal with threading.

Contrary to what you might be used to, in reactive programming, you can use a Thread to process several asynchronous sequences that run at roughly the same time (actually, in non-blocking locksteps). The execution can also easily and often jump from one thread to another.

This arrangement is especially hard for developers that use features dependent on the threading model being more “stable,” such as ThreadLocal. As it lets you associate data with a thread, it becomes tricky to use in a reactive context. As a result, libraries that rely on ThreadLocal at least introduce new challenges when used with Reactor. At worst, they work badly or even fail. Using the MDC of Logback to store and log correlation IDs is a prime example of such a situation.

The usual workaround for ThreadLocal usage is to move the contextual data, C, along your business data, T, in the sequence, by using (for instance) Tuple2<T, C>. This does not look good and leaks an orthogonal concern (the contextual data) into your method and Flux signatures.

Since version 3.1.0, Reactor comes with an advanced feature that is somewhat comparable to ThreadLocal but can be applied to a Flux or a Mono instead of a Thread. This feature is called Context.

As an illustration of what it looks like, the following example both reads from and writes to Context:

In the following sections, we cover Context and how to use it, so that you can eventually understand the preceding example.

This operator can be used when one needs to capture ThreadLocal value(s) at subscription time and reflect these values in the Reactor Context for the benefit of upstream operators. It relies on the context-propagation library and notably the registered ThreadLocalAccessor(s) to discover relevant ThreadLocal values.

This is a convenient alternative to contextWrite which uses the context-propagation API to obtain a ContextSnapshot and then uses that snapshot to populate the Reactor Context.

As a result, if there were any ThreadLocal values during subscription phase, for which there is a registered ThreadLocalAccessor, their values would now be stored in the Reactor Context and visible at runtime in upstream operators.

In very specific cases, your application may deal with types that necessitate some form of cleanup once they are no longer in use. This is an advanced scenario — for, example when you have reference-counted objects or when you deal with off-heap objects. Netty’s ByteBuf is a prime example of both.

In order to ensure proper cleanup of such objects, you need to account for it on a Flux-by-Flux basis, as well as in several of the global hooks (see Using Global Hooks):

This is needed because each hook is made with a specific subset of cleanup in mind, and users might want (for example) to implement specific error-handling logic in addition to cleanup logic within onOperatorError.

Note that some operators are less adapted to dealing with objects that need cleanup. For example, bufferWhen can introduce overlapping buffers, and that means that the discard “local hook” we used earlier might see a first buffer as being discarded and cleanup an element in it that is in a second buffer, where it is still valid.

Although Java does not allow expressing null-safety with its type system, Reactor now provides annotations to declare nullability of APIs, similar to those provided by Spring Framework 5.

Reactor uses these annotations, but they can also be used in any Reactor-based Java project to declare null-safe APIs. Nullability of the types used inside method bodies is outside of the scope of this feature.

These annotations are meta-annotated with JSR 305 annotations (a dormant JSR that is supported by tools such as IntelliJ IDEA) to provide useful warnings to Java developers related to null-safety in order to avoid NullPointerException at runtime. JSR 305 meta-annotations let tooling vendors provide null safety support in a generic way, without having to hard-code support for Reactor annotations.

They are also used by Kotlin, which natively supports null safety. See this dedicated section for more details.

The following annotations are provided in the reactor.util.annotation package:

Suggest Edit to "Advanced Features and Concepts"