消息
1. JMS的
这jakarta.jms.ConnectionFactoryinterface 提供了一种创建jakarta.jms.Connection用于与 JMS 代理交互。
尽管 Spring 需要一个ConnectionFactory要使用 JMS，您通常不需要自己直接使用它，而是可以依赖更高级别的消息传递抽象。
（有关详细信息，请参阅 Spring Framework 参考文档的相关部分。
Spring Boot 还自动配置了发送和接收消息所需的基础设施。
1.1. ActiveMQ “Classic” 支持
当 ActiveMQ “Classic” 在 Classpath 上可用时， Spring Boot 可以配置ConnectionFactory.
如果您使用spring-boot-starter-activemq，提供了连接到 ActiveMQ “Classic” 实例所需的依赖项，以及与 JMS 集成的 Spring 基础架构。 |
ActiveMQ “Classic” 配置由spring.activemq.*.
默认情况下，ActiveMQ “Classic” 自动配置为使用 TCP 传输，默认情况下连接到tcp://localhost:61616.以下示例显示如何更改默认代理 URL：
spring.activemq.broker-url=tcp://192.168.1.210:9876
spring.activemq.user=admin
spring.activemq.password=secretspring:
activemq:
broker-url: "tcp://192.168.1.210:9876"
user: "admin"
password: "secret"默认情况下，CachingConnectionFactory包装本机ConnectionFactory具有合理的设置，您可以通过spring.jms.*:
spring.jms.cache.session-cache-size=5spring:
jms:
cache:
session-cache-size: 5如果您更愿意使用原生池，可以通过向org.messaginghub:pooled-jms并配置JmsPoolConnectionFactory因此，如以下示例所示：
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=50spring:
activemq:
pool:
enabled: true
max-connections: 50看ActiveMQProperties了解更多支持的选项。
您还可以注册任意数量的 bean 来实现ActiveMQConnectionFactoryCustomizer以获取更高级的自定义。 |
默认情况下，ActiveMQ “Classic” 会创建一个目标（如果尚不存在），以便根据提供的名称解析目标。
1.2. ActiveMQ Artemis 支持
Spring Boot 可以自动配置ConnectionFactory当它检测到 ActiveMQ Artemis 在 Classpath 上可用时。
如果存在代理，则会自动启动并配置嵌入式代理（除非已明确设置 mode 属性）。
支持的模式包括embedded（明确表示需要一个嵌入式代理，如果代理在 Classpath 上不可用，则应该发生错误）和native（要使用nettytransport 协议）。
配置后者时， Spring Boot 会配置一个ConnectionFactory连接到使用默认设置在本地计算机上运行的代理。
如果您使用spring-boot-starter-artemis，提供了连接到现有 ActiveMQ Artemis 实例所需的依赖项，以及与 JMS 集成的 Spring 基础架构。
添加org.apache.activemq:artemis-jakarta-server添加到您的应用程序中，允许您使用嵌入式模式。 |
ActiveMQ Artemis 配置由spring.artemis.*.
例如，您可以在application.properties:
spring.artemis.mode=native
spring.artemis.broker-url=tcp://192.168.1.210:9876
spring.artemis.user=admin
spring.artemis.password=secretspring:
artemis:
mode: native
broker-url: "tcp://192.168.1.210:9876"
user: "admin"
password: "secret"在嵌入代理时，您可以选择是否要启用持久性并列出应可用的目标。
这些可以指定为逗号分隔的列表，以便使用默认选项创建它们，或者您可以定义 bean 类型的org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration或org.apache.activemq.artemis.jms.server.config.TopicConfiguration，分别用于高级队列和主题配置。
默认情况下，CachingConnectionFactory包装本机ConnectionFactory具有合理的设置，您可以通过spring.jms.*:
spring.jms.cache.session-cache-size=5spring:
jms:
cache:
session-cache-size: 5如果您更愿意使用本机池，可以通过添加对org.messaginghub:pooled-jms并配置JmsPoolConnectionFactory因此，如以下示例所示：
spring.artemis.pool.enabled=true
spring.artemis.pool.max-connections=50spring:
artemis:
pool:
enabled: true
max-connections: 50看ArtemisProperties以获取更多支持的选项。
不涉及 JNDI 查找，目标是根据其名称解析的，使用name属性或通过配置提供的名称。
1.3. 使用 JNDI ConnectionFactory
如果您在应用程序服务器中运行应用程序，则 Spring Boot 会尝试查找 JMSConnectionFactory通过使用 JNDI。
默认情况下，java:/JmsXA和java:/XAConnectionFactorylocation 进行检查。
您可以使用spring.jms.jndi-nameproperty （如果需要指定备用位置），如以下示例所示：
spring.jms.jndi-name=java:/MyConnectionFactoryspring:
jms:
jndi-name: "java:/MyConnectionFactory"1.4. 发送消息
Spring的JmsTemplate是自动配置的，你可以将其直接自动连接到你自己的 bean 中，如以下示例所示：
@Component
public class MyBean {
private final JmsTemplate jmsTemplate;
public MyBean(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
}
Kotlin
@Component
class MyBean(private val 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.5. 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 jakarta.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:
Java
@Component
public class MyBean {
@JmsListener(destination = "someQueue")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@JmsListener(destination = "someQueue")
fun processMessage(content: String?) {
// ...
}
}
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:
Java
@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 ...
}
}
Kotlin
@Configuration(proxyBeanMethods = false)
class MyJmsConfiguration {
@Bean
fun myFactory(configurer: DefaultJmsListenerContainerFactoryConfigurer): DefaultJmsListenerContainerFactory {
val factory = DefaultJmsListenerContainerFactory()
val connectionFactory = getCustomConnectionFactory()
configurer.configure(factory, connectionFactory)
factory.setMessageConverter(MyMessageConverter())
return factory
}
fun getCustomConnectionFactory() : ConnectionFactory? {
return ...
}
}
Then you can use the factory in any @JmsListener-annotated method as follows:
Java
@Component
public class MyBean {
@JmsListener(destination = "someQueue", containerFactory = "myFactory")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@JmsListener(destination = "someQueue", containerFactory = "myFactory")
fun processMessage(content: String?) {
// ...
}
}
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:
Properties
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=secret
Yaml
spring:
rabbitmq:
host: "localhost"
port: 5672
username: "admin"
password: "secret"
Alternatively, you could configure the same connection using the addresses attribute:
Properties
spring.rabbitmq.addresses=amqp://admin:secret@localhost
Yaml
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.
To make an application-wide, additive customization to the RabbitTemplate, use a RabbitTemplateCustomizer bean.
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:
Java
@Component
public class MyBean {
private final AmqpAdmin amqpAdmin;
private final AmqpTemplate amqpTemplate;
public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
this.amqpAdmin = amqpAdmin;
this.amqpTemplate = amqpTemplate;
}
}
Kotlin
@Component
class MyBean(private val amqpAdmin: AmqpAdmin, private val 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):
Properties
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=2s
Yaml
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:
Properties
spring.rabbitmq.stream.name=my-stream
Yaml
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:
Java
@Component
public class MyBean {
@RabbitListener(queues = "someQueue")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@RabbitListener(queues = ["someQueue"])
fun processMessage(content: String?) {
// ...
}
}
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:
Java
@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 ...
}
}
Kotlin
@Configuration(proxyBeanMethods = false)
class MyRabbitConfiguration {
@Bean
fun myFactory(configurer: SimpleRabbitListenerContainerFactoryConfigurer): SimpleRabbitListenerContainerFactory {
val factory = SimpleRabbitListenerContainerFactory()
val connectionFactory = getCustomConnectionFactory()
configurer.configure(factory, connectionFactory)
factory.setMessageConverter(MyMessageConverter())
return factory
}
fun getCustomConnectionFactory() : ConnectionFactory? {
return ...
}
}
Then you can use the factory in any @RabbitListener-annotated method, as follows:
Java
@Component
public class MyBean {
@RabbitListener(queues = "someQueue", containerFactory = "myFactory")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@RabbitListener(queues = ["someQueue"], containerFactory = "myFactory")
fun processMessage(content: String?) {
// ...
}
}
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:
Properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup
Yaml
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:
Java
@Component
public class MyBean {
private final KafkaTemplate<String, String> kafkaTemplate;
public MyBean(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
}
Kotlin
@Component
class MyBean(private val kafkaTemplate: KafkaTemplate<String, String>) {
}
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:
Java
@Component
public class MyBean {
@KafkaListener(topics = "someTopic")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@KafkaListener(topics = ["someTopic"])
fun processMessage(content: String?) {
// ...
}
}
If a KafkaTransactionManager bean is defined, it is automatically associated to the container factory.
Similarly, if a RecordFilterStrategy, 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 Additional Kafka Properties for more information.
To use the factory bean, wire StreamsBuilder into your @Bean as shown in the following example:
Java
@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(Locale.getDefault()));
}
}
Kotlin
@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
class MyKafkaStreamsConfiguration {
@Bean
fun kStream(streamsBuilder: StreamsBuilder): KStream<Int, String> {
val stream = streamsBuilder.stream<Int, String>("ks1In")
stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(), JsonSerde()))
return stream
}
private fun uppercaseValue(key: Int, value: String): KeyValue<Int?, String?> {
return KeyValue(key, value.uppercase())
}
}
By default, the streams managed by the StreamBuilder object 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 the “Integration Properties” section of the Appendix.
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.
Properties that don’t include a client type (producer, consumer, admin, or streams) in their name are considered to be common and apply to all clients.
Most of these common properties can be overridden for one or more of the client types, if needed.
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 individual client types with additional properties that are not directly supported, use the following properties:
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
Yaml
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, admins, and streams), 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:
Properties
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
Yaml
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:
Properties
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties[spring.json.add.type.headers]=false
Yaml
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:
Java
static {
System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
}
Kotlin
init {
System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers")
}
-
Configure a property name on the @EmbeddedKafka annotation:
Java
@SpringBootTest
@EmbeddedKafka(topics = "someTopic", bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {
// ...
}
Kotlin
@SpringBootTest
@EmbeddedKafka(topics = ["someTopic"], bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {
// ...
}
-
Use a placeholder in configuration properties:
Properties
spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}
Yaml
spring:
kafka:
bootstrap-servers: "${spring.embedded.kafka.brokers}"
4. Apache Pulsar Support
Apache Pulsar is supported by providing auto-configuration of the Spring for Apache Pulsar project.
Spring Boot will auto-configure and register the classic (imperative) Spring for Apache Pulsar components when org.springframework.pulsar:spring-pulsar is on the classpath.
It will do the same for the reactive components when org.springframework.pulsar:spring-pulsar-reactive is on the classpath.
There are spring-boot-starter-pulsar and spring-boot-starter-pulsar-reactive “Starters” for conveniently collecting the dependencies for imperative and reactive use, respectively.
4.1. Connecting to Pulsar
When you use the Pulsar starter, Spring Boot will auto-configure and register a PulsarClient bean.
By default, the application tries to connect to a local Pulsar instance at pulsar://localhost:6650.
This can be adjusted by setting the spring.pulsar.client.service-url property to a different value.
The value must be a valid Pulsar Protocol URL
You can configure the client by specifying any of the spring.pulsar.client.* prefixed application properties.
If you need more control over the configuration, consider registering one or more PulsarClientBuilderCustomizer beans.
4.1.1. Authentication
To connect to a Pulsar cluster that requires authentication, you need to specify which authentication plugin to use by setting the pluginClassName and any parameters required by the plugin.
You can set the parameters as a map of parameter names to parameter values.
The following example shows how to configure the AuthenticationOAuth2 plugin.
Properties
spring.pulsar.client.authentication.plugin-class-name=org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
spring.pulsar.client.authentication.param[issuerUrl]=https://auth.server.cloud/
spring.pulsar.client.authentication.param[privateKey]=file:///Users/some-key.json
spring.pulsar.client.authentication.param.audience=urn:sn:acme:dev:my-instance
Yaml
spring:
pulsar:
client:
authentication:
plugin-class-name: org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
param:
issuerUrl: https://auth.server.cloud/
privateKey: file:///Users/some-key.json
audience: urn:sn:acme:dev:my-instance
You need to ensure that names defined under spring.pulsar.client.authentication.param.* exactly match those expected by your auth plugin (which is typically camel cased).
Spring Boot will not attempt any kind of relaxed binding for these entries.
For example, if you want to configure the issuer url for the AuthenticationOAuth2 auth plugin you must use spring.pulsar.client.authentication.param.issuerUrl.
If you use other forms, such as issuerurl or issuer-url, the setting will not be applied to the plugin.
This lack of relaxed binding also makes using environment variables for authentication parameters problematic because the case sensitivity is lost during translation.
If you use environment variables for the parameters then you will need to follow these steps in the Spring for Apache Pulsar reference documentation for it to work properly.
4.1.2. SSL
By default, Pulsar clients communicate with Pulsar services in plain text.
You can follow these steps in the Spring for Apache Pulsar reference documentation to enable TLS encryption.
For complete details on the client and authentication see the Spring for Apache Pulsar reference documentation.
4.2. Connecting to Pulsar Reactively
When the Reactive auto-configuration is activated, Spring Boot will auto-configure and register a ReactivePulsarClient bean.
The ReactivePulsarClient adapts an instance of the previously described PulsarClient.
Therefore, follow the previous section to configure the PulsarClient used by the ReactivePulsarClient.
4.3. Connecting to Pulsar Administration
Spring for Apache Pulsar’s PulsarAdministration client is also auto-configured.
By default, the application tries to connect to a local Pulsar instance at http://localhost:8080.
This can be adjusted by setting the spring.pulsar.admin.service-url property to a different value in the form (http|https)://<host>:<port>.
If you need more control over the configuration, consider registering one or more PulsarAdminBuilderCustomizer beans.
4.3.1. Authentication
When accessing a Pulsar cluster that requires authentication, the admin client requires the same security configuration as the regular Pulsar client.
You can use the aforementioned authentication configuration by replacing spring.pulsar.client.authentication with spring.pulsar.admin.authentication.
To create a topic on startup, add a bean of type PulsarTopic.
If the topic already exists, the bean is ignored.
4.4. Sending a Message
Spring’s PulsarTemplate is auto-configured, and you can use it to send messages, as shown in the following example:
Java
@Component
public class MyBean {
private final PulsarTemplate<String> pulsarTemplate;
public MyBean(PulsarTemplate<String> pulsarTemplate) {
this.pulsarTemplate = pulsarTemplate;
}
public void someMethod() throws PulsarClientException {
this.pulsarTemplate.send("someTopic", "Hello");
}
}
Kotlin
@Component
class MyBean(private val pulsarTemplate: PulsarTemplate<String>) {
@Throws(PulsarClientException::class)
fun someMethod() {
pulsarTemplate.send("someTopic", "Hello")
}
}
The PulsarTemplate relies on a PulsarProducerFactory to create the underlying Pulsar producer.
Spring Boot auto-configuration also provides this producer factory, which by default, caches the producers that it creates.
You can configure the producer factory and cache settings by specifying any of the spring.pulsar.producer.* and spring.pulsar.producer.cache.* prefixed application properties.
If you need more control over the producer factory configuration, consider registering one or more ProducerBuilderCustomizer beans.
These customizers are applied to all created producers.
You can also pass in a ProducerBuilderCustomizer when sending a message to only affect the current producer.
If you need more control over the message being sent, you can pass in a TypedMessageBuilderCustomizer when sending a message.
4.5. Sending a Message Reactively
When the Reactive auto-configuration is activated, Spring’s ReactivePulsarTemplate is auto-configured, and you can use it to send messages, as shown in the following example:
Java
@Component
public class MyBean {
private final ReactivePulsarTemplate<String> pulsarTemplate;
public MyBean(ReactivePulsarTemplate<String> pulsarTemplate) {
this.pulsarTemplate = pulsarTemplate;
}
public void someMethod() {
this.pulsarTemplate.send("someTopic", "Hello").subscribe();
}
}
Kotlin
@Component
class MyBean(private val pulsarTemplate: ReactivePulsarTemplate<String>) {
fun someMethod() {
pulsarTemplate.send("someTopic", "Hello").subscribe()
}
}
The ReactivePulsarTemplate relies on a ReactivePulsarSenderFactory to actually create the underlying sender.
Spring Boot auto-configuration also provides this sender factory, which by default, caches the producers that it creates.
You can configure the sender factory and cache settings by specifying any of the spring.pulsar.producer.* and spring.pulsar.producer.cache.* prefixed application properties.
If you need more control over the sender factory configuration, consider registering one or more ReactiveMessageSenderBuilderCustomizer beans.
These customizers are applied to all created senders.
You can also pass in a ReactiveMessageSenderBuilderCustomizer when sending a message to only affect the current sender.
If you need more control over the message being sent, you can pass in a MessageSpecBuilderCustomizer when sending a message.
4.6. Receiving a Message
When the Apache Pulsar infrastructure is present, any bean can be annotated with @PulsarListener to create a listener endpoint.
The following component creates a listener endpoint on the someTopic topic:
Java
@Component
public class MyBean {
@PulsarListener(topics = "someTopic")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@PulsarListener(topics = ["someTopic"])
fun processMessage(content: String?) {
// ...
}
}
Spring Boot auto-configuration provides all the components necessary for PulsarListener, such as the PulsarListenerContainerFactory and the consumer factory it uses to construct the underlying Pulsar consumers.
You can configure these components by specifying any of the spring.pulsar.listener.* and spring.pulsar.consumer.* prefixed application properties.
If you need more control over the consumer factory configuration, consider registering one or more ConsumerBuilderCustomizer beans.
These customizers are applied to all consumers created by the factory, and therefore all @PulsarListener instances.
You can also customize a single listener by setting the consumerCustomizer attribute of the @PulsarListener annotation.
4.7. Receiving a Message Reactively
When the Apache Pulsar infrastructure is present and the Reactive auto-configuration is activated, any bean can be annotated with @ReactivePulsarListener to create a reactive listener endpoint.
The following component creates a reactive listener endpoint on the someTopic topic:
Java
@Component
public class MyBean {
@ReactivePulsarListener(topics = "someTopic")
public Mono<Void> processMessage(String content) {
// ...
return Mono.empty();
}
}
Kotlin
@Component
class MyBean {
@ReactivePulsarListener(topics = ["someTopic"])
fun processMessage(content: String?): Mono<Void> {
// ...
return Mono.empty()
}
}
Spring Boot auto-configuration provides all the components necessary for ReactivePulsarListener, such as the ReactivePulsarListenerContainerFactory and the consumer factory it uses to construct the underlying reactive Pulsar consumers.
You can configure these components by specifying any of the spring.pulsar.listener. and spring.pulsar.consumer. prefixed application properties.
If you need more control over the consumer factory configuration, consider registering one or more ReactiveMessageConsumerBuilderCustomizer beans.
These customizers are applied to all consumers created by the factory, and therefore all @ReactivePulsarListener instances.
You can also customize a single listener by setting the consumerCustomizer attribute of the @ReactivePulsarListener annotation.
4.8. Reading a Message
The Pulsar reader interface enables applications to manually manage cursors.
When you use a reader to connect to a topic you need to specify which message the reader begins reading from when it connects to a topic.
When the Apache Pulsar infrastructure is present, any bean can be annotated with @PulsarReader to consume messages using a reader.
The following component creates a reader endpoint that starts reading messages from the beginning of the someTopic topic:
Java
@Component
public class MyBean {
@PulsarReader(topics = "someTopic", startMessageId = "earliest")
public void processMessage(String content) {
// ...
}
}
Kotlin
@Component
class MyBean {
@PulsarReader(topics = ["someTopic"], startMessageId = "earliest")
fun processMessage(content: String?) {
// ...
}
}
The @PulsarReader relies on a PulsarReaderFactory to create the underlying Pulsar reader.
Spring Boot auto-configuration provides this reader factory which can be customized by setting any of the spring.pulsar.reader.* prefixed application properties.
If you need more control over the reader factory configuration, consider registering one or more ReaderBuilderCustomizer beans.
These customizers are applied to all readers created by the factory, and therefore all @PulsarReader instances.
You can also customize a single listener by setting the readerCustomizer attribute of the @PulsarReader annotation.
4.9. Reading a Message Reactively
When the Apache Pulsar infrastructure is present and the Reactive auto-configuration is activated, Spring’s ReactivePulsarReaderFactory is provided, and you can use it to create a reader in order to read messages in a reactive fashion.
The following component creates a reader using the provided factory and reads a single message from 5 minutes ago from the someTopic topic:
Java
@Component
public class MyBean {
private final ReactivePulsarReaderFactory<String> pulsarReaderFactory;
public MyBean(ReactivePulsarReaderFactory<String> pulsarReaderFactory) {
this.pulsarReaderFactory = pulsarReaderFactory;
}
public void someMethod() {
ReactiveMessageReaderBuilderCustomizer<String> readerBuilderCustomizer = (readerBuilder) -> readerBuilder
.topic("someTopic")
.startAtSpec(StartAtSpec.ofInstant(Instant.now().minusSeconds(5)));
Mono<Message<String>> message = this.pulsarReaderFactory
.createReader(Schema.STRING, List.of(readerBuilderCustomizer))
.readOne();
// ...
}
}
Kotlin
@Component
class MyBean(private val pulsarReaderFactory: ReactivePulsarReaderFactory<String>) {
fun someMethod() {
val readerBuilderCustomizer = ReactiveMessageReaderBuilderCustomizer {
readerBuilder: ReactiveMessageReaderBuilder<String> ->
readerBuilder
.topic("someTopic")
.startAtSpec(StartAtSpec.ofInstant(Instant.now().minusSeconds(5)))
}
val message = pulsarReaderFactory
.createReader(Schema.STRING, listOf(readerBuilderCustomizer))
.readOne()
// ...
}
}
Spring Boot auto-configuration provides this reader factory which can be customized by setting any of the spring.pulsar.reader.* prefixed application properties.
If you need more control over the reader factory configuration, consider passing in one or more ReactiveMessageReaderBuilderCustomizer instances when using the factory to create a reader.
If you need more control over the reader factory configuration, consider registering one or more ReactiveMessageReaderBuilderCustomizer beans.
These customizers are applied to all created readers.
You can also pass one or more ReactiveMessageReaderBuilderCustomizer when creating a reader to only apply the customizations to the created reader.
For more details on any of the above components and to discover other available features, see the Spring for Apache Pulsar reference documentation.
4.10. Additional Pulsar Properties
The properties supported by auto-configuration are shown in the “Integration Properties” section of the Appendix.
Note that, for the most part, these properties (hyphenated or camelCase) map directly to the Apache Pulsar configuration properties.
See the Apache Pulsar documentation for details.
Only a subset of the properties supported by Pulsar are available directly through the PulsarProperties class.
If you wish to tune the auto-configured components with additional properties that are not directly supported, you can use the customizer supported by each aforementioned component.
5. 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.
5.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.
5.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:
Properties
spring.rsocket.server.mapping-path=/rsocket
spring.rsocket.server.transport=websocket
Yaml
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:
Properties
spring.rsocket.server.port=9898
Yaml
spring:
rsocket:
server:
port: 9898
5.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.
5.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:
Java
@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);
}
}
Kotlin
@Service
class MyService(rsocketRequesterBuilder: RSocketRequester.Builder) {
private val rsocketRequester: RSocketRequester
init {
rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898)
}
fun someRSocketCall(name: String): Mono<User> {
return rsocketRequester.route("user").data(name).retrieveMono(
User::class.java
)
}
}
6. 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:
Properties
spring.integration.jdbc.initialize-schema=always
Yaml
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:
Properties
# Connecting to a RSocket server over TCP
spring.integration.rsocket.client.host=example.org
spring.integration.rsocket.client.port=9898
Yaml
# Connecting to a RSocket server over TCP
spring:
integration:
rsocket:
client:
host: "example.org"
port: 9898
Properties
# Connecting to a RSocket Server over WebSocket
spring.integration.rsocket.client.uri=ws://example.org
Yaml
# Connecting to a RSocket Server over WebSocket
spring:
integration:
rsocket:
client:
uri: "ws://example.org"
See the IntegrationAutoConfiguration and IntegrationProperties classes for more details.
7. WebSockets
Spring Boot provides WebSockets auto-configuration for embedded Tomcat, Jetty, and Undertow.
If you deploy a war file to a standalone container, Spring Boot assumes that the container is responsible for the configuration of its WebSocket support.
Spring Framework provides rich WebSocket support for MVC web applications that can be easily accessed through the spring-boot-starter-websocket module.
WebSocket support is also available for reactive web applications and requires to include the WebSocket API alongside spring-boot-starter-webflux:
<dependency>
<groupId>jakarta.websocket</groupId>
<artifactId>jakarta.websocket-api</artifactId>
</dependency>
8. 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.