Messaging
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’ and Spring Boot also provides auto-configuration
options for RabbitTemplate
and RabbitMQ. There is also support for STOMP messaging
natively in Spring WebSocket and Spring Boot has support for that through starters and a
small amount of auto-configuration. Spring Boot also has support for Apache Kafka.
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 won’t need to use it directly yourself
and you 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.
ActiveMQ support
Spring Boot can also configure a ConnectionFactory
when it detects that ActiveMQ is
available on the classpath. If the broker is present, an embedded broker is started and
configured automatically (as long as no broker URL is specified through configuration).
If you are using spring-boot-starter-activemq the necessary dependencies to
connect or embed an ActiveMQ instance are provided, as well as the Spring infrastructure
to integrate with JMS.
|
ActiveMQ configuration is controlled by external configuration properties in
spring.activemq.*
. For example, you might declare the following section in
application.properties
:
spring.activemq.broker-url=tcp://192.168.1.210:9876
spring.activemq.user=admin
spring.activemq.password=secret
You can also pool JMS resources by adding a dependency to
org.apache.activemq:activemq-pool
and configure the PooledConnectionFactory
accordingly:
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=50
See
ActiveMQProperties
for more of the supported options. You can also register an arbitrary number of beans
implementing ActiveMQConnectionFactoryCustomizer for more advanced customizations.
|
By default, ActiveMQ creates a destination if it does not exist yet, so destinations are resolved against their provided names.
Artemis support
Spring Boot can auto-configure a ConnectionFactory
when it detects that Artemis is
available on the classpath. If the broker is present, an embedded broker is started and
configured automatically (unless the mode property has been explicitly set). The supported
modes are: embedded
(to make explicit that an embedded broker is required and should
lead to an error if the broker is not available in the classpath), and native
to connect
to a broker using the netty
transport protocol. When the latter is configured, Spring
Boot configures a ConnectionFactory
connecting to a broker running on the local machine
with the default settings.
If you are using spring-boot-starter-artemis the necessary dependencies to
connect to an existing Artemis instance are provided, as well as the Spring infrastructure
to integrate with JMS. Adding org.apache.activemq:artemis-jms-server to your application
allows you to use the embedded mode.
|
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.host=192.168.1.210
spring.artemis.port=9876
spring.artemis.user=admin
spring.artemis.password=secret
When embedding the broker, you can choose if you want to enable persistence, and the list
of 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.
See
ArtemisProperties
for more of the supported options.
No JNDI lookup is involved at all and destinations are resolved against their names, either using the ‘name’ attribute in the Artemis configuration or the names provided through configuration.
Using a JNDI ConnectionFactory
If you are running your application in an Application Server Spring Boot will attempt to
locate a JMS ConnectionFactory
using JNDI. By default the locations java:/JmsXA
and
java:/XAConnectionFactory
will be checked. You can use the
spring.jms.jndi-name
property if you need to specify an alternative location:
spring.jms.jndi-name=java:/MyConnectionFactory
Sending a message
Spring’s JmsTemplate
is auto-configured and you can autowire it directly into your own
beans:
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;
@Component
public class MyBean {
private final JmsTemplate jmsTemplate;
@Autowired
public MyBean(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
// ...
}
JmsMessagingTemplate
can be injected in a similar manner. If a DestinationResolver or MessageConverter
beans are defined, they are associated automatically to the auto-configured
JmsTemplate .
|
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
or MessageConverter
beans are defined, they are associated automatically to the default factory.
The default factory is transactional by default. If you are running in an infrastructure
where a JtaTransactionManager
is present, it will be associated to the listener container
by default. If not, the sessionTransacted
flag will be 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
will make sure 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) {
// ...
}
}
Check 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 exposes another factory that uses a specific
MessageConverter
:
@Configuration
static class JmsConfiguration {
@Bean
public DefaultJmsListenerContainerFactory myFactory(
DefaultJmsListenerContainerFactoryConfigurer configurer) {
DefaultJmsListenerContainerFactory factory =
new DefaultJmsListenerContainerFactory();
configurer.configure(factory, connectionFactory());
factory.setMessageConverter(myMessageConverter());
return factory;
}
}
Then you can use in any @JmsListener
-annotated method as follows:
@Component
public class MyBean {
@JmsListener(destination = "someQueue", containerFactory="myFactory")
public void processMessage(String content) {
// ...
}
}
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 via RabbitMQ, including the
spring-boot-starter-amqp
‘Starter’.
RabbitMQ support
RabbitMQ is a lightweight, reliable, scalable and portable message broker based on the
AMQP protocol. Spring uses RabbitMQ
to communicate using 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
See RabbitProperties
for more of the supported options.
Check Understanding AMQP, the protocol used by RabbitMQ for more details. |
Sending a message
Spring’s AmqpTemplate
and AmqpAdmin
are auto-configured and you can autowire them
directly into your own beans:
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class MyBean {
private final AmqpAdmin amqpAdmin;
private final AmqpTemplate amqpTemplate;
@Autowired
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 .
|
Any org.springframework.amqp.core.Queue
that is defined as a bean will be automatically
used to declare a corresponding queue on the RabbitMQ instance if necessary.
You can enable retries on the AmqpTemplate
to retry operations, for example in the event
the broker connection is lost. Retries are disabled by default.
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 one is configured automatically. If a MessageConverter
or
MessageRecoverer
beans are defined, they are associated automatically to the default
factory.
The following component creates a listener endpoint on the someQueue
queue:
@Component
public class MyBean {
@RabbitListener(queues = "someQueue")
public void processMessage(String content) {
// ...
}
}
Check 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
that you can use to initialize a
SimpleRabbitListenerContainerFactory
with the same settings as the one that is
auto-configured.
For instance, the following exposes another factory that uses a specific
MessageConverter
:
@Configuration
static class RabbitConfiguration {
@Bean
public SimpleRabbitListenerContainerFactory myFactory(
SimpleRabbitListenerContainerFactoryConfigurer configurer) {
SimpleRabbitListenerContainerFactory factory =
new SimpleRabbitListenerContainerFactory();
configurer.configure(factory, connectionFactory);
factory.setMessageConverter(myMessageConverter());
return factory;
}
}
Then you can use 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 will be rejected and either dropped
or routed to a dead-letter exchange if the broker is configured so. Retries are disabled
by default.
If retries are not enabled and the listener throws an exception, by default the
delivery will be retried indefinitely. You can modify this behavior in two ways; set the
defaultRequeueRejected property to false and zero re-deliveries will be attempted; or,
throw an AmqpRejectAndDontRequeueException to signal the message should be rejected.
This is the mechanism used when retries are enabled and the maximum delivery attempts are
reached.
|
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
See KafkaProperties
for more of the supported options.
Sending a Message
Spring’s KafkaTemplate
is auto-configured and you can autowire them directly in your own
beans:
@Component
public class MyBean {
private final KafkaTemplate kafkaTemplate;
@Autowired
public MyBean(KafkaTemplate kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
// ...
}
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 configured automatically 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) {
// ...
}
}
Additional Kafka Properties
The properties supported by auto configuration are shown in [common-application-properties]. Note that these properties (hyphenated or camelCase) map directly to the Apache Kafka dotted properties for the most part, refer to the Apache Kafka documentation for details.
The first few of these properties apply to both producers and consumers, but can be specified at the producer or consumer level if you wish to use different values for each. Apache Kafka designates properties with an importance: HIGH, MEDIUM and LOW. Spring Boot auto configuration supports all HIGH importance properties, some selected MEDIUM and LOW, and any that do not have a default value.
Only a subset of the properties supported by Kafka are available via the KafkaProperties
class. If you wish to configure the producer or consumer with additional properties that
are not directly supported, use the following:
spring.kafka.properties.foo.bar=baz
This sets the common foo.bar
Kafka property to baz
.
These properties will be shared by both the consumer and producer factory beans. If you wish to customize these components with different properties, such as to use a different metrics reader for each, you can override the bean definitions, as follows:
@Configuration
public static class CustomKafkaBeans {
/**
* Customized ProducerFactory bean.
* @param properties the kafka properties.
* @return the bean.
*/
@Bean
public ProducerFactory<?, ?> kafkaProducerFactory(KafkaProperties properties) {
Map<String, Object> producerProperties = properties.buildProducerProperties();
producerProperties.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
MyProducerMetricsReporter.class);
return new DefaultKafkaProducerFactory<Object, Object>(producerProperties);
}
/**
* Customized ConsumerFactory bean.
* @param properties the kafka properties.
* @return the bean.
*/
@Bean
public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties) {
Map<String, Object> consumerProperties = properties.buildConsumerProperties();
consumerProperties.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
MyConsumerMetricsReporter.class);
return new DefaultKafkaConsumerFactory<Object, Object>(consumerProperties);
}
}