The Spring Framework provides extensive support for integrating with messaging systems, from simplified use of the JMS API using JmsTemplate
to a complete infrastructure to receive messages asynchronously.
Spring AMQP provides a similar feature set for the Advanced Message Queuing Protocol.
Spring Boot also provides auto-configuration options for RabbitTemplate
and RabbitMQ.
Spring WebSocket natively includes support for STOMP messaging, and Spring Boot has support for that through starters and a small amount of auto-configuration.
Spring Boot also has support for Apache Kafka.
1. JMS
The javax.jms.ConnectionFactory
interface provides a standard method of creating a javax.jms.Connection
for interacting with a JMS broker.
Although Spring needs a ConnectionFactory
to work with JMS, you generally need not use it directly yourself and can instead rely on higher level messaging abstractions.
(See the relevant section of the Spring Framework reference documentation for details.)
Spring Boot also auto-configures the necessary infrastructure to send and receive messages.
1.1. ActiveMQ Artemis Support
Spring Boot can auto-configure a ConnectionFactory
when it detects that ActiveMQ Artemis is available on the classpath.
If the broker is present, an embedded broker is automatically started and configured (unless the mode property has been explicitly set).
The supported modes are embedded
(to make explicit that an embedded broker is required and that an error should occur if the broker is not available on the classpath) and native
(to connect to a broker using the netty
transport protocol).
When the latter is configured, Spring Boot configures a ConnectionFactory
that connects to a broker running on the local machine with the default settings.
If you use spring-boot-starter-artemis , the necessary dependencies to connect to an existing ActiveMQ Artemis instance are provided, as well as the Spring infrastructure to integrate with JMS.
Adding org.apache.activemq:artemis-jms-server to your application lets you use embedded mode.
|
ActiveMQ Artemis configuration is controlled by external configuration properties in spring.artemis.*
.
For example, you might declare the following section in application.properties
:
spring.artemis.mode=native
spring.artemis.broker-url=tcp://192.168.1.210:9876
spring.artemis.user=admin
spring.artemis.password=secret
spring:
artemis:
mode: native
broker-url: "tcp://192.168.1.210:9876"
user: "admin"
password: "secret"
When embedding the broker, you can choose if you want to enable persistence and list the destinations that should be made available.
These can be specified as a comma-separated list to create them with the default options, or you can define bean(s) of type org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration
or org.apache.activemq.artemis.jms.server.config.TopicConfiguration
, for advanced queue and topic configurations, respectively.
By default, a CachingConnectionFactory
wraps the native ConnectionFactory
with sensible settings that you can control by external configuration properties in spring.jms.*
:
spring.jms.cache.session-cache-size=5
spring:
jms:
cache:
session-cache-size: 5
No JNDI lookup is involved, and destinations are resolved against their names, using either the name
attribute in the Artemis configuration or the names provided through configuration.
1.2. Using a JNDI ConnectionFactory
If you are running your application in an application server, Spring Boot tries to locate a JMS ConnectionFactory
by using JNDI.
By default, the java:/JmsXA
and java:/XAConnectionFactory
location are checked.
You can use the spring.jms.jndi-name
property if you need to specify an alternative location, as shown in the following example:
spring.jms.jndi-name=java:/MyConnectionFactory
spring:
jms:
jndi-name: "java:/MyConnectionFactory"
1.3. Sending a Message
Spring’s JmsTemplate
is auto-configured, and you can autowire it directly into your own beans, as shown in the following example:
@Component
public class MyBean {
private final JmsTemplate jmsTemplate;
public MyBean(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
}
JmsMessagingTemplate can be injected in a similar manner.
If a DestinationResolver or a MessageConverter bean is defined, it is associated automatically to the auto-configured JmsTemplate .
|
1.4. Receiving a Message
When the JMS infrastructure is present, any bean can be annotated with @JmsListener
to create a listener endpoint.
If no JmsListenerContainerFactory
has been defined, a default one is configured automatically.
If a DestinationResolver
, a MessageConverter
, or a javax.jms.ExceptionListener
beans are defined, they are associated automatically with the default factory.
By default, the default factory is transactional.
If you run in an infrastructure where a JtaTransactionManager
is present, it is associated to the listener container by default.
If not, the sessionTransacted
flag is enabled.
In that latter scenario, you can associate your local data store transaction to the processing of an incoming message by adding @Transactional
on your listener method (or a delegate thereof).
This ensures that the incoming message is acknowledged, once the local transaction has completed.
This also includes sending response messages that have been performed on the same JMS session.
The following component creates a listener endpoint on the someQueue
destination:
@Component
public class MyBean {
@JmsListener(destination = "someQueue")
public void processMessage(String content) {
// ...
}
}
See the Javadoc of @EnableJms for more details.
|
If you need to create more JmsListenerContainerFactory
instances or if you want to override the default, Spring Boot provides a DefaultJmsListenerContainerFactoryConfigurer
that you can use to initialize a DefaultJmsListenerContainerFactory
with the same settings as the one that is auto-configured.
For instance, the following example exposes another factory that uses a specific MessageConverter
:
@Configuration(proxyBeanMethods = false)
public class MyJmsConfiguration {
@Bean
public DefaultJmsListenerContainerFactory myFactory(DefaultJmsListenerContainerFactoryConfigurer configurer) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
ConnectionFactory connectionFactory = getCustomConnectionFactory();
configurer.configure(factory, connectionFactory);
factory.setMessageConverter(new MyMessageConverter());
return factory;
}
private ConnectionFactory getCustomConnectionFactory() {
return ...
}
}
Then you can use the factory in any @JmsListener
-annotated method as follows:
@Component
public class MyBean {
@JmsListener(destination = "someQueue", containerFactory = "myFactory")
public void processMessage(String content) {
// ...
}
}
2. AMQP
The Advanced Message Queuing Protocol (AMQP) is a platform-neutral, wire-level protocol for message-oriented middleware.
The Spring AMQP project applies core Spring concepts to the development of AMQP-based messaging solutions.
Spring Boot offers several conveniences for working with AMQP through RabbitMQ, including the spring-boot-starter-amqp
“Starter”.
2.1. RabbitMQ support
RabbitMQ is a lightweight, reliable, scalable, and portable message broker based on the AMQP protocol.
Spring uses RabbitMQ
to communicate through the AMQP protocol.
RabbitMQ configuration is controlled by external configuration properties in spring.rabbitmq.*
.
For example, you might declare the following section in application.properties
:
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=secret
spring:
rabbitmq:
host: "localhost"
port: 5672
username: "admin"
password: "secret"
Alternatively, you could configure the same connection using the addresses
attribute:
spring.rabbitmq.addresses=amqp://admin:secret@localhost
spring:
rabbitmq:
addresses: "amqp://admin:secret@localhost"
When specifying addresses that way, the host and port properties are ignored.
If the address uses the amqps protocol, SSL support is enabled automatically.
|
See RabbitProperties
for more of the supported property-based configuration options.
To configure lower-level details of the RabbitMQ ConnectionFactory
that is used by Spring AMQP, define a ConnectionFactoryCustomizer
bean.
If a ConnectionNameStrategy
bean exists in the context, it will be automatically used to name connections created by the auto-configured CachingConnectionFactory
.
See Understanding AMQP, the protocol used by RabbitMQ for more details. |
2.2. Sending a Message
Spring’s AmqpTemplate
and AmqpAdmin
are auto-configured, and you can autowire them directly into your own beans, as shown in the following example:
@Component
public class MyBean {
private final AmqpAdmin amqpAdmin;
private final AmqpTemplate amqpTemplate;
public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
this.amqpAdmin = amqpAdmin;
this.amqpTemplate = amqpTemplate;
}
}
RabbitMessagingTemplate can be injected in a similar manner.
If a MessageConverter bean is defined, it is associated automatically to the auto-configured AmqpTemplate .
|
If necessary, any org.springframework.amqp.core.Queue
that is defined as a bean is automatically used to declare a corresponding queue on the RabbitMQ instance.
To retry operations, you can enable retries on the AmqpTemplate
(for example, in the event that the broker connection is lost):
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=2s
spring:
rabbitmq:
template:
retry:
enabled: true
initial-interval: "2s"
Retries are disabled by default.
You can also customize the RetryTemplate
programmatically by declaring a RabbitRetryTemplateCustomizer
bean.
If you need to create more RabbitTemplate
instances or if you want to override the default, Spring Boot provides a RabbitTemplateConfigurer
bean that you can use to initialize a RabbitTemplate
with the same settings as the factories used by the auto-configuration.
2.3. Sending a Message To A Stream
To send a message to a particular stream, specify the name of the stream, as shown in the following example:
spring.rabbitmq.stream.name=my-stream
spring:
rabbitmq:
stream:
name: "my-stream"
If a MessageConverter
, StreamMessageConverter
, or ProducerCustomizer
bean is defined, it is associated automatically to the auto-configured RabbitStreamTemplate
.
If you need to create more RabbitStreamTemplate
instances or if you want to override the default, Spring Boot provides a RabbitStreamTemplateConfigurer
bean that you can use to initialize a RabbitStreamTemplate
with the same settings as the factories used by the auto-configuration.
2.4. Receiving a Message
When the Rabbit infrastructure is present, any bean can be annotated with @RabbitListener
to create a listener endpoint.
If no RabbitListenerContainerFactory
has been defined, a default SimpleRabbitListenerContainerFactory
is automatically configured and you can switch to a direct container using the spring.rabbitmq.listener.type
property.
If a MessageConverter
or a MessageRecoverer
bean is defined, it is automatically associated with the default factory.
The following sample component creates a listener endpoint on the someQueue
queue:
@Component
public class MyBean {
@RabbitListener(queues = "someQueue")
public void processMessage(String content) {
// ...
}
}
See the Javadoc of @EnableRabbit for more details.
|
If you need to create more RabbitListenerContainerFactory
instances or if you want to override the default, Spring Boot provides a SimpleRabbitListenerContainerFactoryConfigurer
and a DirectRabbitListenerContainerFactoryConfigurer
that you can use to initialize a SimpleRabbitListenerContainerFactory
and a DirectRabbitListenerContainerFactory
with the same settings as the factories used by the auto-configuration.
It does not matter which container type you chose. Those two beans are exposed by the auto-configuration. |
For instance, the following configuration class exposes another factory that uses a specific MessageConverter
:
@Configuration(proxyBeanMethods = false)
public class MyRabbitConfiguration {
@Bean
public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
ConnectionFactory connectionFactory = getCustomConnectionFactory();
configurer.configure(factory, connectionFactory);
factory.setMessageConverter(new MyMessageConverter());
return factory;
}
private ConnectionFactory getCustomConnectionFactory() {
return ...
}
}
Then you can use the factory in any @RabbitListener
-annotated method, as follows:
@Component
public class MyBean {
@RabbitListener(queues = "someQueue", containerFactory = "myFactory")
public void processMessage(String content) {
// ...
}
}
You can enable retries to handle situations where your listener throws an exception.
By default, RejectAndDontRequeueRecoverer
is used, but you can define a MessageRecoverer
of your own.
When retries are exhausted, the message is rejected and either dropped or routed to a dead-letter exchange if the broker is configured to do so.
By default, retries are disabled.
You can also customize the RetryTemplate
programmatically by declaring a RabbitRetryTemplateCustomizer
bean.
By default, if retries are disabled and the listener throws an exception, the delivery is retried indefinitely.
You can modify this behavior in two ways: Set the defaultRequeueRejected property to false so that zero re-deliveries are attempted or throw an AmqpRejectAndDontRequeueException to signal the message should be rejected.
The latter is the mechanism used when retries are enabled and the maximum number of delivery attempts is reached.
|
3. Apache Kafka Support
Apache Kafka is supported by providing auto-configuration of the spring-kafka
project.
Kafka configuration is controlled by external configuration properties in spring.kafka.*
.
For example, you might declare the following section in application.properties
:
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup
spring:
kafka:
bootstrap-servers: "localhost:9092"
consumer:
group-id: "myGroup"
To create a topic on startup, add a bean of type NewTopic .
If the topic already exists, the bean is ignored.
|
See KafkaProperties
for more supported options.
3.1. Sending a Message
Spring’s KafkaTemplate
is auto-configured, and you can autowire it directly in your own beans, as shown in the following example:
@Component
public class MyBean {
private final KafkaTemplate<String, String> kafkaTemplate;
public MyBean(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
}
If the property spring.kafka.producer.transaction-id-prefix is defined, a KafkaTransactionManager is automatically configured.
Also, if a RecordMessageConverter bean is defined, it is automatically associated to the auto-configured KafkaTemplate .
|
3.2. Receiving a Message
When the Apache Kafka infrastructure is present, any bean can be annotated with @KafkaListener
to create a listener endpoint.
If no KafkaListenerContainerFactory
has been defined, a default one is automatically configured with keys defined in spring.kafka.listener.*
.
The following component creates a listener endpoint on the someTopic
topic:
@Component
public class MyBean {
@KafkaListener(topics = "someTopic")
public void processMessage(String content) {
// ...
}
}
If a KafkaTransactionManager
bean is defined, it is automatically associated to the container factory.
Similarly, if a RecordFilterStrategy
, ErrorHandler
, CommonErrorHandler
, AfterRollbackProcessor
or ConsumerAwareRebalanceListener
bean is defined, it is automatically associated to the default factory.
Depending on the listener type, a RecordMessageConverter
or BatchMessageConverter
bean is associated to the default factory.
If only a RecordMessageConverter
bean is present for a batch listener, it is wrapped in a BatchMessageConverter
.
A custom ChainedKafkaTransactionManager must be marked @Primary as it usually references the auto-configured KafkaTransactionManager bean.
|
3.3. Kafka Streams
Spring for Apache Kafka provides a factory bean to create a StreamsBuilder
object and manage the lifecycle of its streams.
Spring Boot auto-configures the required KafkaStreamsConfiguration
bean as long as kafka-streams
is on the classpath and Kafka Streams is enabled by the @EnableKafkaStreams
annotation.
Enabling Kafka Streams means that the application id and bootstrap servers must be set.
The former can be configured using spring.kafka.streams.application-id
, defaulting to spring.application.name
if not set.
The latter can be set globally or specifically overridden only for streams.
Several additional properties are available using dedicated properties; other arbitrary Kafka properties can be set using the spring.kafka.streams.properties
namespace.
See also features.html for more information.
To use the factory bean, wire StreamsBuilder
into your @Bean
as shown in the following example:
@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
public class MyKafkaStreamsConfiguration {
@Bean
public KStream<Integer, String> kStream(StreamsBuilder streamsBuilder) {
KStream<Integer, String> stream = streamsBuilder.stream("ks1In");
stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(), new JsonSerde<>()));
return stream;
}
private KeyValue<Integer, String> uppercaseValue(Integer key, String value) {
return new KeyValue<>(key, value.toUpperCase());
}
}
By default, the streams managed by the StreamBuilder
object it creates are started automatically.
You can customize this behavior using the spring.kafka.streams.auto-startup
property.
3.4. Additional Kafka Properties
The properties supported by auto configuration are shown in application-properties.html. Note that, for the most part, these properties (hyphenated or camelCase) map directly to the Apache Kafka dotted properties. See the Apache Kafka documentation for details.
The first few of these properties apply to all components (producers, consumers, admins, and streams) but can be specified at the component level if you wish to use different values. Apache Kafka designates properties with an importance of HIGH, MEDIUM, or LOW. Spring Boot auto-configuration supports all HIGH importance properties, some selected MEDIUM and LOW properties, and any properties that do not have a default value.
Only a subset of the properties supported by Kafka are available directly through the KafkaProperties
class.
If you wish to configure the producer or consumer with additional properties that are not directly supported, use the following properties:
spring.kafka.properties[prop.one]=first
spring.kafka.admin.properties[prop.two]=second
spring.kafka.consumer.properties[prop.three]=third
spring.kafka.producer.properties[prop.four]=fourth
spring.kafka.streams.properties[prop.five]=fifth
spring:
kafka:
properties:
"[prop.one]": "first"
admin:
properties:
"[prop.two]": "second"
consumer:
properties:
"[prop.three]": "third"
producer:
properties:
"[prop.four]": "fourth"
streams:
properties:
"[prop.five]": "fifth"
This sets the common prop.one
Kafka property to first
(applies to producers, consumers and admins), the prop.two
admin property to second
, the prop.three
consumer property to third
, the prop.four
producer property to fourth
and the prop.five
streams property to fifth
.
You can also configure the Spring Kafka JsonDeserializer
as follows:
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties[spring.json.value.default.type]=com.example.Invoice
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.main,com.example.another
spring:
kafka:
consumer:
value-deserializer: "org.springframework.kafka.support.serializer.JsonDeserializer"
properties:
"[spring.json.value.default.type]": "com.example.Invoice"
"[spring.json.trusted.packages]": "com.example.main,com.example.another"
Similarly, you can disable the JsonSerializer
default behavior of sending type information in headers:
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties[spring.json.add.type.headers]=false
spring:
kafka:
producer:
value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"
properties:
"[spring.json.add.type.headers]": false
Properties set in this way override any configuration item that Spring Boot explicitly supports. |
3.5. Testing with Embedded Kafka
Spring for Apache Kafka provides a convenient way to test projects with an embedded Apache Kafka broker.
To use this feature, annotate a test class with @EmbeddedKafka
from the spring-kafka-test
module.
For more information, please see the Spring for Apache Kafka reference manual.
To make Spring Boot auto-configuration work with the aforementioned embedded Apache Kafka broker, you need to remap a system property for embedded broker addresses (populated by the EmbeddedKafkaBroker
) into the Spring Boot configuration property for Apache Kafka.
There are several ways to do that:
-
Provide a system property to map embedded broker addresses into
spring.kafka.bootstrap-servers
in the test class:
static {
System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
}
-
Configure a property name on the
@EmbeddedKafka
annotation:
@SpringBootTest
@EmbeddedKafka(topics = "someTopic", bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {
// ...
}
-
Use a placeholder in configuration properties:
spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}
spring:
kafka:
bootstrap-servers: "${spring.embedded.kafka.brokers}"
4. RSocket
RSocket is a binary protocol for use on byte stream transports. It enables symmetric interaction models through async message passing over a single connection.
The spring-messaging
module of the Spring Framework provides support for RSocket requesters and responders, both on the client and on the server side.
See the RSocket section of the Spring Framework reference for more details, including an overview of the RSocket protocol.
4.1. RSocket Strategies Auto-configuration
Spring Boot auto-configures an RSocketStrategies
bean that provides all the required infrastructure for encoding and decoding RSocket payloads.
By default, the auto-configuration will try to configure the following (in order):
-
CBOR codecs with Jackson
-
JSON codecs with Jackson
The spring-boot-starter-rsocket
starter provides both dependencies.
See the Jackson support section to know more about customization possibilities.
Developers can customize the RSocketStrategies
component by creating beans that implement the RSocketStrategiesCustomizer
interface.
Note that their @Order
is important, as it determines the order of codecs.
4.2. RSocket server Auto-configuration
Spring Boot provides RSocket server auto-configuration.
The required dependencies are provided by the spring-boot-starter-rsocket
.
Spring Boot allows exposing RSocket over WebSocket from a WebFlux server, or standing up an independent RSocket server. This depends on the type of application and its configuration.
For WebFlux application (that is of type WebApplicationType.REACTIVE
), the RSocket server will be plugged into the Web Server only if the following properties match:
spring.rsocket.server.mapping-path=/rsocket
spring.rsocket.server.transport=websocket
spring:
rsocket:
server:
mapping-path: "/rsocket"
transport: "websocket"
Plugging RSocket into a web server is only supported with Reactor Netty, as RSocket itself is built with that library. |
Alternatively, an RSocket TCP or websocket server is started as an independent, embedded server. Besides the dependency requirements, the only required configuration is to define a port for that server:
spring.rsocket.server.port=9898
spring:
rsocket:
server:
port: 9898
4.3. Spring Messaging RSocket support
Spring Boot will auto-configure the Spring Messaging infrastructure for RSocket.
This means that Spring Boot will create a RSocketMessageHandler
bean that will handle RSocket requests to your application.
4.4. Calling RSocket Services with RSocketRequester
Once the RSocket
channel is established between server and client, any party can send or receive requests to the other.
As a server, you can get injected with an RSocketRequester
instance on any handler method of an RSocket @Controller
.
As a client, you need to configure and establish an RSocket connection first.
Spring Boot auto-configures an RSocketRequester.Builder
for such cases with the expected codecs and applies any RSocketConnectorConfigurer
bean.
The RSocketRequester.Builder
instance is a prototype bean, meaning each injection point will provide you with a new instance .
This is done on purpose since this builder is stateful and you should not create requesters with different setups using the same instance.
The following code shows a typical example:
@Service
public class MyService {
private final RSocketRequester rsocketRequester;
public MyService(RSocketRequester.Builder rsocketRequesterBuilder) {
this.rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898);
}
public Mono<User> someRSocketCall(String name) {
return this.rsocketRequester.route("user").data(name).retrieveMono(User.class);
}
}
5. Spring Integration
Spring Boot offers several conveniences for working with Spring Integration, including the spring-boot-starter-integration
“Starter”.
Spring Integration provides abstractions over messaging and also other transports such as HTTP, TCP, and others.
If Spring Integration is available on your classpath, it is initialized through the @EnableIntegration
annotation.
Spring Integration polling logic relies on the auto-configured TaskScheduler
.
The default PollerMetadata
(poll unbounded number of messages every second) can be customized with spring.integration.poller.*
configuration properties.
Spring Boot also configures some features that are triggered by the presence of additional Spring Integration modules.
If spring-integration-jmx
is also on the classpath, message processing statistics are published over JMX.
If spring-integration-jdbc
is available, the default database schema can be created on startup, as shown in the following line:
spring.integration.jdbc.initialize-schema=always
spring:
integration:
jdbc:
initialize-schema: "always"
If spring-integration-rsocket
is available, developers can configure an RSocket server using "spring.rsocket.server.*"
properties and let it use IntegrationRSocketEndpoint
or RSocketOutboundGateway
components to handle incoming RSocket messages.
This infrastructure can handle Spring Integration RSocket channel adapters and @MessageMapping
handlers (given "spring.integration.rsocket.server.message-mapping-enabled"
is configured).
Spring Boot can also auto-configure an ClientRSocketConnector
using configuration properties:
# Connecting to a RSocket server over TCP
spring.integration.rsocket.client.host=example.org
spring.integration.rsocket.client.port=9898
# Connecting to a RSocket server over TCP
spring:
integration:
rsocket:
client:
host: "example.org"
port: 9898
# Connecting to a RSocket Server over WebSocket
spring.integration.rsocket.client.uri=ws://example.org
# Connecting to a RSocket Server over WebSocket
spring:
integration:
rsocket:
client:
uri: "ws://example.org"
See the IntegrationAutoConfiguration
and IntegrationProperties
classes for more details.
6. What to Read Next
The next section describes how to enable IO capabilities in your application. You can read about caching, mail, validation, rest clients and more in this section.