For non-reactive applications, RabbitMQ Java Client provides the most complete API to manage resources, publish messages to and consume messages from RabbitMQ. Note Reactor RabbitMQ is based on RabbitMQ Java Client.

Applications using RabbitMQ as a message bus using this API may consider switching to Reactor RabbitMQ if the application is implemented in a functional style.

Spring AMQP applies core Spring Framework concepts to the development of AMQP-based messaging solutions. It provides a "template" as a high-level abstraction for sending and receiving messages. It also provides support for Message-driven POJOs with a "listener container". These libraries facilitate management of AMQP resources while promoting the use of dependency injection and declarative configuration. Spring AMQP is based on RabbitMQ Java Client.

Start RabbitMQ on your local machine with all the defaults (e.g. AMQP port is 5672).

Download Reactor RabbitMQ from github.com/reactor/reactor-rabbitmq/.

To build your own application using the Reactor RabbitMQ API, you need to include a dependency to Reactor RabbitMQ.

For Gradle:

For Maven:

When using a milestone or a release candidate, you need to add the Spring IO milestone repository.

For Gradle:

For Maven:

When using a snapshot, you need to add the Spring IO snapshots repository.

For Gradle:

For Maven:

Reactor RabbitMQ used semantic versioning from version 1.0 to version 1.4, but switched to another scheme for consistency with Reactor Core and the other Reactor libraries.

Starting from 1.4, Reactor RabbitMQ uses a GENERATION.MAJOR.MINOR scheme, whereby an increment in:

The Sender is also able to declare and delete AMQP resources the reactive way. You can learn more about the AMQP model on RabbitMQ website.

Sender has a declare* method for each type of resource (exchange, binding, and queue) and there’s also a respective *Specification class to describe each creation.

Note the Sender#declare* methods return their respective AMQP results wrapped into a Mono.

One can also use the ResourcesSpecification factory class with a static import to reduce boilerplate code. Combined with Mono chaining and Sender#declare shortcuts, it allows for condensed syntax:

Sender has delete* and delete methods as well. Here is an example with the short method forms:

Sender offers also the sendWithPublishConfirms method to send messages and receive publisher confirms to make sure the broker has taken into account the outbound messages.

Sender#sendWithPublishConfirms returns a Flux<OutboundMessageResult> that can be subscribed to to know that outbound messages have successfully reached the broker.

It is also possible to know about unroutable messages, that is messages not routed to any queue because they do not match any routing rule. Tracking of unroutable messages is disabled by default, it can be enabled by using the Sender#sendWithPublishConfirms(Publisher<OutboundMessage>, SendOptions) method and set the trackReturned flag on SendOptions:

The OutboundMessageResult#isReturned method then tells whether a message has been routed somewhere or not. This method always returns false if unroutable messages are not tracked. Note the OutboundMessageResult#isAck method returns true for unroutable messages, because the broker considered they have been taken care of (i.e. confirmed). So if you are interested in unroutable messages, the returned status should always be checked before the confirmed status.

Note it is possible to publish OutboundMessage instances of a custom class and get them back in the flux of OutboundMessageResult. To do so, use the sendWithTypedPublishConfirms method:

The previous sample uses the provided CorrelableOutboundMessage class, but it could be any subclass of OutboundMessage. The OutboundMessageResult instances of the confirmation flux are typed with the original message class, so the extra information is available. This allows to perform any useful processing on this extra information when the outbound message is confirmed.

Reactor RabbitMQ configure by default the Java Client to use NIO, i.e. there’s only one thread that deals with IO. This can be changed by specifying a ConnectionFactory in the SenderOptions.

The Sender uses 2 Reactor’s Scheduler: one for the subscription when creating the connection and another one for resources management. The Sender defaults to 2 bounded elastic schedulers, this can be overriden in the SenderOptions. The Sender takes care of disposing the default schedulers when closing. If not using the default schedulers, it’s developer’s job to dispose schedulers they passed in to the SenderOptions.

When the Sender is no longer required, the instance can be closed. The underlying Connection is closed, as well as the default schedulers if none has been explicitly provided.

The send and sendWithPublishConfirms methods can take an additional SendOptions parameter to specify the behavior to adopt if the publishing of a message fails. The default behavior is to retry every 200 milliseconds for 10 seconds in case of connection failure. As automatic connection recovery is enabled by default, the connection is likely to be re-opened after a network glitch and the flux of outbound messages should stall only during connection recovery before restarting automatically. This default behavior tries to find a trade-off between reactivity and robustness.

You can customize the retry by settings your own instance of RetrySendingExceptionHandler in the SendOptions, e.g. to retry for 20 seconds every 500 milliseconds:

The RetrySendingExceptionHandler uses a Predicate<Throwable> to decide whether an exception should trigger a retry or not. If the exception isn’t retryable, the exception handler wraps the exception in a RabbitFluxException and throws it.

For consistency sake, the retry exception handler used with ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE (the default) will trigger retry attempts for the same conditions as connection recovery triggering. This means that if connection recovery has kicked in, publishing will be retried at least for the retry timeout configured (10 seconds by default).

Note the exception handler is a BiConsumer<Sender.SendContext, Exception>, where Sender.SendContext is a class providing access to the OutboundMessage and the underlying AMQP Channel. This makes it easy to customize the default behavior: logging BiConsumer#andThen retrying, only logging, trying to send the message somewhere else, etc.

Reactor RabbitMQ supports reactive request/reply. From RabbitMQ documentation:

For performance reason, Reactor RabbitMQ builds on top direct reply-to. The next snippet shows the usage of the RpcClient class:

In the example above, a consumer waits on the rpc.server.queue to process requests. A RpcClient is created from a Sender, it will send requests to a given exchange with a given routing key. The RpcClient handles the machinery to send the request and wait on a reply queue the result processed on the server queue, wrapping everything up with reactive API. Note a RPC client isn’t meant to be used for only 1 request, it can be a long-lived object handling different requests, as long as they’re directed to the same destination (defined by the exchange and the routing key passed in when the RpcClient is created).

A RpcClient uses a sequence of Long for correlation, but this can be changed by passing in a Supplier<String> when creating the RpcClient:

This can be useful e.g. when the RPC server can make sense of the correlation ID.

The Receiver class has different flavors of the receive* method and each of them can accept a ConsumeOptions instance. Here are the different options:

Receiver has several receive* methods that differ on the way consumer are acknowledged back to the broker. Acknowledgment mode can have profound impacts on performance and memory consumption.

To learn more on how the ConsumeOptions#qos setting can impact the behavior of Receiver#consumeAutoAck and Receiver#consumeManualAck, have a look at this post about queuing theory.

When the Receiver is no longer required, the instance can be closed. The underlying Connection is closed, as well as the default scheduler if none has been explicitly provided.

Network connection between the broker and the client can fail. This is transparent for consumers thanks to RabbitMQ Java client automatic connection recovery. Connection failures affect sending though, and acknowledgment is a sending operation.

When using Receiver#consumeAutoAck, acknowledgments are retried for 10 seconds every 200 milliseconds in case of connection failure. This can be changed by setting the BiConsumer<Receiver.AcknowledgmentContext, Exception> exceptionHandler in the ConsumeOptions, e.g. to retry for 20 seconds every 500 milliseconds:

When using Receiver#consumeManualAck, acknowledgment is handled by the developer, who can do pretty anything they want on acknowledgment failure.

AcknowledgableDelivery#ack and AcknowledgableDelivery#nack methods handle retry internally based on BiConsumer<Receiver.AcknowledgmentContext, Exception> exceptionHandler in the ConsumeOptions. Developer does not have to execute retry explicitly on acknowledgment failure and benefits from Reactor RabbitMQ retry support when acknowledging a message:

Note the exception handler is a BiConsumer<Receiver.AcknowledgmentContext, Exception>. This means acknowledgment failure can be handled in any way, here we choose to retry the acknowledgment. Note also that by using ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE, we choose to retry only on unexpected connection failures and rely on the AMQP Java client to automatically re-create a new connection in the background. The decision to retry on a given exception can be customized by providing a Predicate<Throwable> in place of ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE.

It is possible to specify only a ConnectionFactory for Sender/ReceiverOptions and let Reactor RabbitMQ create connection from this ConnectionFactory. Internally, Reactor RabbitMQ will create a Mono<Connection> to perform its operations and the connection will be created only when needed.

When the developer lets Reactor RabbitMQ create a Mono<Connection>, the library will take responsibility for the following actions for each instance of Sender and Receiver:

Reactor RabbitMQ provides 2 ways to have more control over the connection creation, e.g. to provide a name or to connect to different nodes:

Sender and Receiver instances create their own Connection but it’s possible to use only one or a few Connection instances to be able to use exclusive resources between a Sender and a Receiver or simply to control the number of created connections.

Both SenderOptions and ReceiverOptions have a connectionSupplier method that can encapsulate any logic to create the Connection the Sender or Receiver will end up using through a Mono<Connection>. Reactor RabbitMQ provides a way to share the exact same connection instance between some Sender and Receiver instances:

Be aware that closing the first Sender or Receiver will close the underlying AMQP connection for all the others.

SenderOptions provides a channelMono property that is called when creating the Channel used in sending methods. This is a convenient way to provide any custom logic when creating the Channel, e.g. retry logic.

A Sender instance maintains a Mono<Channel> to manage resources and by default the underlying Channel is cached. A new Channel is also automatically created in case of error. Channel creation is not a cheap operation, so this default behavior fits most use cases. Each resource management method provides a counterpart method with an additional ResourceManagementOptions argument. This allows to provide a custom Mono<Channel> for a given resource operation. This can be useful when multiple threads are using the same Sender instance, to avoid using the same Channel from multiple threads.

In the example above, each operation will use the same Channel as it is cached. This way these operations won’t interfer with any other thread using the default resource management Mono<Channel> in the Sender instance.

By default, Sender#send* methods open a new Channel for every call. This is OK for long-running calls, e.g. when the flux of outbound messages is infinite. For workloads whereby Sender#send* is called often for finite, short flux of messages, opening a new Channel every time may not be optimal.

It is possible to use a pool of channels as part of the SendOptions when sending outbound messages with Sender, as illustrated in the following snippet:

Note it is developer’s responsibility to close the pool when it is no longer necessary, typically at application shutdown.

Micro-benchmarks revealed channel pooling performs much better for sending short sequence of messages (1 to 10) repeatedly, without publisher confirms. With longer sequence of messages (100 or more), channel pooling can perform worse than without pooling at all. According to the same micro-benchmarks, channel pooling does not make sending with publisher confirms perform better, it appears to perform even worse. Don’t take these conclusions for granted, you should always make your own benchmarks depending on your workloads.

Some applications may want to fail fast and throw an exception when the broker is unavailable. Other applications may want to retry connection opening when it fails.

Both SenderOptions and ReceiverOptions provide a Function<Mono<? extends Connection>, Mono<? extends Connection>> connectionMonoConfigurator, which is a hook in the Mono<Connection> creation of the Sender or Receiver instance. This is a good place to customize the Mono<Connection> to configure a retry policy.

The following snippet shows how to configure retry with an inline connectionMonoConfigurator for a Sender:

Please read the Reactor Core documentation for more information about retry, retry with exponential backoff, and retry support in Reactor-Extra.