Messaging

1. JMS

javax.jms.ConnectionFactory接口提供了创建用于与JMS代理交互的javax.jms.Connection的标准方法。尽管Spring需要ConnectionFactory来使用JMS，但您通常不需要直接使用它，而是可以依赖更高级别的消息传递抽象。(有关详细信息，请参阅Spring框架参考文档的相关部分。)Spring Boot还自动配置发送和接收消息所需的基础设施。

1.1. ActiveMQ Artemis Support

当Spring Boot检测到类路径上有ActiveMQ Artemis时，它可以自动配置ConnectionFactory。如果存在代理，则会自动启动和配置嵌入的代理(除非显式设置了模式属性)。支持的模式是Embedded(明确表示需要嵌入式代理，如果类路径上没有可用的代理，则会出现错误)和原生(使用netty传输协议连接到代理)。当配置后者时，Spring Boot使用默认设置配置一个ConnectionFactory，它连接到在本地机器上运行的代理。

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-jakarta-server to your application lets you use embedded mode.

ActiveMQ Artemis配置由spring.artemis.*中的外部配置属性控制。例如，您可以在Applation.Properties中声明以下部分：

Properties

Yaml

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"

在嵌入代理时，您可以选择是否要启用持久性，并列出应该可用的目的地。可以将它们指定为逗号分隔的列表，以使用默认选项创建它们，或者您可以分别为高级队列和主题配置定义org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration或org.apache.activemq.artemis.jms.server.config.TopicConfiguration，类型的Bean。

默认情况下，CachingConnectionFactory使用您可以通过spring.jms.*中的外部配置属性控制的合理设置包装本机ConnectionFactory：

Properties

Yaml

spring.jms.cache.session-cache-size=5

spring:
  jms:
    cache:
      session-cache-size: 5

如果您更愿意使用原生池，可以通过添加对org.MessagingHub：pooled-jms的依赖并相应地配置JmsPoolConnectionFactory来实现，如下例所示：

Properties

Yaml

spring.artemis.pool.enabled=true
spring.artemis.pool.max-connections=50

spring:
  artemis:
    pool:
      enabled: true
      max-connections: 50

有关更多支持的选项，请参阅ArtemisProperties。

不涉及JNDI查找，使用Artemis配置中的name属性或通过配置提供的名称解析目的地。

1.2. Using a JNDI ConnectionFactory

如果您在应用程序服务器上运行应用程序，Spring Boot会尝试使用JNDI定位JMSConnectionFactory。默认情况下，选中Java：/JmsXA和Java：/XAConnectionFactory位置。如果需要指定备用位置，可以使用spring.jms.jndi-name属性，如下例所示：

Properties

Yaml

spring.jms.jndi-name=java:/MyConnectionFactory

spring:
  jms:
    jndi-name: "java:/MyConnectionFactory"

1.3. Sending a Message

Spring的JmsTemplate是自动配置的，您可以直接将其自动绑定到您自己的Bean中，如下例所示：

Java

Kotlin

import org.springframework.jms.core.JmsTemplate; import org.springframework.stereotype.Component; @Component public class MyBean { private final JmsTemplate jmsTemplate; public MyBean(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; }  // ...  public void someMethod() { this.jmsTemplate.convertAndSend("hello"); } }

import org.springframework.jms.core.JmsTemplate import org.springframework.stereotype.Component @Component class MyBean(private val jmsTemplate: JmsTemplate) {  // ...  fun someMethod() { jmsTemplate.convertAndSend("hello") } }

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

当存在JMS基础设施时，任何Bean都可以用@JmsListener注释来创建侦听器端点。如果没有定义JmsListenerContainerFactory，则会自动配置一个默认的。如果定义了DestinationResolver、MessageConverter或javax.jms.ExceptionListenerBean，它们将自动与默认工厂关联。

默认情况下，默认工厂是事务性的。如果您在存在JtaTransactionManager的基础设施中运行，则默认情况下，它与侦听器容器相关联。如果不是，则启用sessionTransated标志。在后一种情况下，通过在侦听器方法(或其委托)上添加@Transaction，可以将本地数据存储事务与传入消息处理相关联。这确保了在本地事务完成后确认传入的消息。这还包括发送已在同一JMS会话上执行的响应消息。

以下组件在omeQueue目标上创建侦听器终结点：

Java

Kotlin

import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; @Component public class MyBean { @JmsListener(destination = "someQueue") public void processMessage(String content) { // ... } }

import org.springframework.jms.annotation.JmsListener import org.springframework.stereotype.Component @Component class MyBean { @JmsListener(destination = "someQueue") fun processMessage(content: String?) { // ... } }

See the Javadoc of @EnableJms for more details.

如果您需要创建更多JmsListenerContainerFactory实例，或者如果您想要覆盖缺省值，Spring Boot提供了一个DefaultJmsListenerContainerFactoryConfigurer，您可以使用它来使用与自动配置相同的设置来初始化DefaultJmsListenerContainerFactory。

例如，下面的示例公开了另一个使用特定MessageConverter的工厂：

Java

Kotlin

import jakarta.jms.ConnectionFactory; import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; @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 ... } }

import jakarta.jms.ConnectionFactory import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.jms.config.DefaultJmsListenerContainerFactory @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 ... } }

然后，您可以在任何@JmsListener注释的方法中使用工厂，如下所示：

Java

Kotlin

import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; @Component public class MyBean { @JmsListener(destination = "someQueue", containerFactory = "myFactory") public void processMessage(String content) { // ... } }

import org.springframework.jms.annotation.JmsListener import org.springframework.stereotype.Component @Component class MyBean { @JmsListener(destination = "someQueue", containerFactory = "myFactory") fun processMessage(content: String?) { // ... } }

2. AMQP

高级消息队列协议(AMQP)是面向消息中间件的平台中立、线路级协议。Spring AMQP项目将核心的Spring概念应用于基于AMQP的消息传递解决方案的开发。Spring Boot为通过RabbitMQ使用AMQP提供了几个便利，包括Spring-ot-starter-amqp“starter”。

2.1. RabbitMQ Support

RabbitMQ是一个基于AMQP协议的轻量级、可靠、可扩展和可移植的消息代理。Spring使用RabbitMQ通过AMQP协议进行通信。

RabbitMQ配置由spring.rabbitmq.*中的外部配置属性控制。例如，您可以在Applation.Properties中声明以下部分：

Properties

Yaml

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"

或者，您可以使用Addresses属性配置相同的连接：

Properties

Yaml

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.

有关支持的更多属性配置选项，请参阅RabbitProperties。要配置Spring AMQP使用的RabbitMQConnectionFactory的低层详细信息，请定义一个ConnectionFactoryCustomizerBean。

如果上下文中存在ConnectionNameStrategyBean，它将自动用于命名由自动配置的CachingConnectionFactory创建的连接。

See Understanding AMQP, the protocol used by RabbitMQ for more details.

2.2. Sending a Message

Spring的AmqpTemplate和AmqpAdmin是自动配置的，您可以直接将它们自动绑定到您自己的Bean中，如下例所示：

Java

Kotlin

import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.stereotype.Component; @Component public class MyBean { private final AmqpAdmin amqpAdmin; private final AmqpTemplate amqpTemplate; public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) { this.amqpAdmin = amqpAdmin; this.amqpTemplate = amqpTemplate; }  // ...  public void someMethod() { this.amqpAdmin.getQueueInfo("someQueue"); } public void someOtherMethod() { this.amqpTemplate.convertAndSend("hello"); } }

import org.springframework.amqp.core.AmqpAdmin import org.springframework.amqp.core.AmqpTemplate import org.springframework.stereotype.Component @Component class MyBean(private val amqpAdmin: AmqpAdmin, private val amqpTemplate: AmqpTemplate) {  // ...  fun someMethod() { amqpAdmin.getQueueInfo("someQueue") } fun someOtherMethod() { amqpTemplate.convertAndSend("hello") } }

RabbitMessagingTemplate can be injected in a similar manner. If a MessageConverter bean is defined, it is associated automatically to the auto-configured AmqpTemplate.

如有必要，定义为org.springframework.amqp.core.Queue的任何Bean将自动用于在RabbitMQ实例上声明相应的队列。

要重试操作，您可以在AmqpTemplate上启用重试(例如，在Broker连接丢失的情况下)：

Properties

Yaml

spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=2s

spring:
  rabbitmq:
    template:
      retry:
        enabled: true
        initial-interval: "2s"

默认情况下禁用重试。您还可以通过声明RabbitRetryTemplateCustomizerBean，以编程方式定制RetryTemplate。

如果您需要创建更多RabbitTemplate实例，或者如果您想覆盖默认设置，Spring Boot提供了一个RabbitTemplateConfigurerBean，您可以使用该Bean来初始化RabbitTemplate，其设置与自动配置使用的工厂相同。

2.3. Sending a Message To A Stream

要将消息发送到特定流，请指定流的名称，如下例所示：

Properties

Yaml

spring.rabbitmq.stream.name=my-stream

spring:
  rabbitmq:
    stream:
      name: "my-stream"

如果定义了MessageConverter、StreamMessageConverter或ProducerCustomizerBean，它会自动关联到自动配置的RabbitStreamTemplate。

如果您需要创建更多的RabbitStreamTemplate实例，或者如果您想覆盖默认设置，Spring Boot提供了一个RabbitStreamTemplateConfigurerBean，您可以使用它来初始化RabbitStreamTemplate，其设置与自动配置使用的工厂相同。

2.4. Receiving a Message

当存在Rabbit基础设施时，任何Bean都可以用@RabbitListener注释来创建侦听器端点。如果没有定义<SimpleRabbitListenerContainerFactory>RabbitListenerContainerFactory，则会自动配置一个默认类型，您可以使用spring.rabbitmq.listener.type属性切换到直接容器。如果定义了MessageConverter或MessageRecovererBean，它将自动与默认工厂关联。

以下示例组件在omeQueue队列上创建侦听器终结点：

Java

Kotlin

import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class MyBean { @RabbitListener(queues = "someQueue") public void processMessage(String content) { // ... } }

import org.springframework.amqp.rabbit.annotation.RabbitListener import org.springframework.stereotype.Component @Component class MyBean { @RabbitListener(queues = ["someQueue"]) fun processMessage(content: String?) { // ... } }

See the Javadoc of @EnableRabbit for more details.

如果您需要创建更多RabbitListenerContainerFactory实例，或者如果您想覆盖默认设置，Spring Boot提供了一个SimpleRabbitListenerContainerFactoryConfigurer和一个DirectRabbitListenerContainerFactoryConfigurer，您可以使用它们来初始化SimpleRabbitListenerContainerFactory和DirectRabbitListenerContainerFactory，它们具有与自动配置使用的工厂相同的设置。

It does not matter which container type you chose. Those two beans are exposed by the auto-configuration.

例如，下面的配置类公开使用特定MessageConverter的另一个工厂：

Java

Kotlin

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @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 ... } }

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory import org.springframework.amqp.rabbit.connection.ConnectionFactory import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration @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 ... } }

然后，您可以在任何@RabbitListener注释的方法中使用工厂，如下所示：

Java

Kotlin

import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class MyBean { @RabbitListener(queues = "someQueue", containerFactory = "myFactory") public void processMessage(String content) { // ... } }

import org.springframework.amqp.rabbit.annotation.RabbitListener import org.springframework.stereotype.Component @Component class MyBean { @RabbitListener(queues = ["someQueue"], containerFactory = "myFactory") fun processMessage(content: String?) { // ... } }

您可以启用重试来处理侦听器引发异常的情况。默认情况下，使用RejectAndDontRequeeRecoverer，但您可以定义自己的MessageRecoverer。当重试次数耗尽时，消息将被拒绝，并被丢弃或路由到死信交换(如果代理被配置为这样做)。默认情况下，重试被禁用。您还可以通过声明RabbitRetryTemplateCustomizerBean，以编程方式定制RetryTemplate。

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

通过提供Spring-kafka工程的自动配置，支持apache kafka。

Kafka配置由spring.kafka.*中的外部配置属性控制。例如，您可以在Applation.Properties中声明以下部分：

Properties

Yaml

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.

有关更多支持的选项，请参阅KafkaProperties。

3.1. Sending a Message

Spring的KafkaTemplate是自动配置的，您可以直接在自己的Bean中自动绑定，如下例所示：

Java

Kotlin

import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Component; @Component public class MyBean { private final KafkaTemplate<String, String> kafkaTemplate; public MyBean(KafkaTemplate<String, String> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; }  // ...  public void someMethod() { this.kafkaTemplate.send("someTopic", "Hello"); } }

import org.springframework.kafka.core.KafkaTemplate import org.springframework.stereotype.Component @Component class MyBean(private val kafkaTemplate: KafkaTemplate<String, String>) {  // ...  fun someMethod() { kafkaTemplate.send("someTopic", "Hello") } }

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

当存在ApacheKafka基础设施时，任何Bean都可以用@KafkaListener注释来创建侦听器端点。如果没有定义KafkaListenerContainerFactory，则会使用spring.kafka.listener.*中定义的key自动配置一个默认密钥。

以下组件在某些主题主题上创建侦听器终结点：

Java

Kotlin

import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @Component public class MyBean { @KafkaListener(topics = "someTopic") public void processMessage(String content) { // ... } }

import org.springframework.kafka.annotation.KafkaListener import org.springframework.stereotype.Component @Component class MyBean { @KafkaListener(topics = ["someTopic"]) fun processMessage(content: String?) { // ... } }

如果定义了KafkaTransactionManagerBean，它将自动关联到容器工厂。同样，如果定义了RecordFilterStrategy、CommonErrorHandler、AfterRollback Processor或Consumer AwareRebalanceListenerBean，它将自动关联到默认工厂。

根据侦听器类型，RecordMessageConverter或BatchMessageConverterBean与默认工厂相关联。如果批处理侦听器只存在RecordMessageConverterBean，则它包装在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提供了一个工厂Bean来创建StreamsBuilder对象并管理其流的生命周期。只要Kafka-Streams在类路径上，并且@EnableKafkaStreams注释启用了Kafka Streams，Spring Boot就会自动配置所需的KafkaStreamsConfigurationBean。

启用Kafka Streams意味着必须设置应用程序ID和引导服务器。前者可以使用spring.kafka.streams.application-id，进行配置，如果未设置，则默认为spring.applation.name。后者可以全局设置，也可以仅为流专门覆盖。

使用专用属性可以使用几个额外的属性；其他任意的Kafka属性可以使用spring.kafka.stres.properties名称空间进行设置。有关详细信息，另请参阅其他卡夫卡属性。

要使用工厂Bean，请将StreamsBuilder连接到您的@Bean，如下例所示：

Java

Kotlin

import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafkaStreams; import org.springframework.kafka.support.serializer.JsonSerde; @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()); } }

import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.KeyValue import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.Produced import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.kafka.annotation.EnableKafkaStreams import org.springframework.kafka.support.serializer.JsonSerde @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()) } }

默认情况下，由StreamBuilder对象管理的流自动启动。您可以使用spring.kafka.streams.auto-startup属性来自定义此行为。

3.4. Additional Kafka Properties

自动配置支持的属性显示在附录的“集成属性”部分。请注意，在大多数情况下，这些属性(带连字符或CamelCase)直接映射到以点分隔的ApacheKafka属性。有关详细信息，请参阅阿帕奇·卡夫卡文档。

这些属性中的前几个适用于所有组件(生产者、消费者、管理员和流)，但如果您希望使用不同的值，则可以在组件级别指定这些属性。阿帕奇·卡夫卡将属性的重要性指定为高、中或低。Spring Boot自动配置支持所有高重要性属性、一些选定的中低属性以及没有缺省值的任何属性。

只有Kafka支持的属性子集可以通过KafkaProperties类直接使用。如果要使用不直接支持的其他属性配置生产者或消费者，请使用以下属性：

Properties

Yaml

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"

这会将常见的prop.onekafka属性设置为First(适用于生产者、消费者和管理员)，将proport.Twoadmin属性设置为Second，将prop.Three消费者属性设置为Third，将pro.fourProducer属性设置为第四，并将pro.FiveStreams属性设置为第五。

您也可以通过如下方式配置Spring KafkaJsonOverializer：

Properties

Yaml

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"

同样，您可以禁用在Header中发送类型信息的JsonSerializer默认行为：

Properties

Yaml

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 ApacheKafka提供了一种使用嵌入的ApacheKafka代理测试项目的便捷方法。要使用此功能，请使用Spring-kafka-test模块中的@EmbeddedKafka注释测试类。有关更多信息，请参阅参考手册。

要使Spring Boot自动配置与前面提到的嵌入式Apache Kafka代理一起工作，您需要将嵌入式代理地址的系统属性(由EmbeddedKafkaBroker填充)重新映射到Apache Kafka的Spring Boot配置属性中。有几种方法可以做到这一点：

提供一个系统属性，将嵌入的代理地址映射到测试类中的spring.kafka.bootstrap-Servers：

Java

Kotlin

static {
    System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
}

init {
    System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers")
}

在@EmbeddedKafka批注上配置属性名称：

Java

Kotlin

import org.springframework.boot.test.context.SpringBootTest; import org.springframework.kafka.test.context.EmbeddedKafka; @SpringBootTest @EmbeddedKafka(topics = "someTopic", bootstrapServersProperty = "spring.kafka.bootstrap-servers") class MyTest { // ... }

import org.springframework.boot.test.context.SpringBootTest import org.springframework.kafka.test.context.EmbeddedKafka @SpringBootTest @EmbeddedKafka(topics = ["someTopic"], bootstrapServersProperty = "spring.kafka.bootstrap-servers") class MyTest { // ... }

在配置属性中使用占位符：

Properties

Yaml

spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}

spring:
  kafka:
    bootstrap-servers: "${spring.embedded.kafka.brokers}"

4. RSocket

RSocket是用于字节流传输的二进制协议。它通过在单个连接上传递异步消息来启用对称交互模型。

Spring框架的Spring-Messaging模块为客户端和服务器端的RSocket请求器和响应器提供支持。有关更多详细信息，包括RSocket协议的概述，请参阅Spring框架参考的RSocket部分。

4.1. RSocket Strategies Auto-configuration

Spring Boot自动配置RSocketStrategyBean，该Bean提供编码和解码RSocket有效负载所需的所有基础设施。默认情况下，自动配置将尝试配置以下内容(按顺序)：

CBOR与Jackson的编解码器
JSON与杰克逊的编解码器

SpringBoot-starter-r套接字starter提供了这两个依赖项。有关定制可能性的更多信息，请参阅Jackson支持部分。

开发人员可以通过创建实现RSocketStrategiesCustomizer接口的Bean来自定义RSocketStrategy组件。请注意，它们的@order很重要，因为它决定了编解码器的顺序。

4.2. RSocket server Auto-configuration

Spring Boot提供RSocket服务器自动配置。所需的依赖项由SpringBoot-starter-r套接字提供。

Spring Boot允许通过WebSocket从WebFlux服务器公开RSocket，或建立一个独立的RSocket服务器。这取决于应用程序的类型及其配置。

对于WebFlux应用程序(即类型WebApplicationType.REACTIVE)，仅当以下属性匹配时，才会将RSocket服务器插入Web服务器：

Properties

Yaml

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.

或者，RSocket TCP或WebSocket服务器作为独立的嵌入式服务器启动。除了相关性要求之外，唯一需要的配置是为该服务器定义一个端口：

Properties

Yaml

spring.rsocket.server.port=9898

spring:
  rsocket:
    server:
      port: 9898

4.3. Spring Messaging RSocket support

Spring Boot将为RSocket自动配置Spring消息传递基础设施。

这意味着Spring Boot将创建一个RSocketMessageHandlerBean来处理对应用程序的RSocket请求。

4.4. Calling RSocket Services with RSocketRequester

在服务器和客户端之间建立RSocket通道后，任何一方都可以向另一方发送或接收请求。

作为服务器，您可以在RSocket@Controller的任何处理程序方法上注入RSocketRequester实例。作为客户端，您需要首先配置和建立RSocket连接。Spring Boot使用预期的编解码器为此类情况自动配置RSocketRequester.Builder，并应用任何RSocketConnectorConfigurerBean。

RSocketRequester.Builder实例是一个原型Bean，这意味着每个注入点都将为您提供一个新实例。这是故意这样做的，因为此构建器是有状态的，您不应该使用同一实例创建具有不同设置的请求器。

下面的代码显示了一个典型的示例：

Java

Kotlin

import reactor.core.publisher.Mono; import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.stereotype.Service; @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); } }

import org.springframework.messaging.rsocket.RSocketRequester import org.springframework.stereotype.Service import reactor.core.publisher.Mono @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 ) } }

5. Spring Integration

Spring Boot为使用Spring Integration提供了几个便利，包括Spring-boot-starter-集成“starter”。Spring集成提供了消息传递和其他传输(如HTTP、TCP和其他传输)的抽象。如果您的类路径上有Spring集成可用，那么它将通过@EnableIntegration注释进行初始化。

Spring集成轮询逻辑依赖于自动配置的TaskScheduler。默认的PollerMetadata(每秒轮询无限数量的消息)可以使用spring.Integration.poller.*配置属性进行自定义。

Spring Boot还配置了一些由附加的Spring集成模块触发的功能。如果类路径上也有Spring-Integration-jmx，则消息处理统计信息将通过JMX发布。如果可以使用Spring-Integration-jdbc，则可以在启动时创建默认的数据库模式，如下面一行所示：

Properties

Yaml

spring.integration.jdbc.initialize-schema=always

spring:
  integration:
    jdbc:
      initialize-schema: "always"

如果有Spring-Integration-r套接字可用，开发人员可以使用“spring.rsocket.server.*”属性配置RSocket服务器，并让它使用IntegrationRSocketEndpoint或RSocketOutound Gateway组件处理传入的RSocket消息。该基础设施可以处理Spring Integration RSocket通道适配器和@Messagemap处理程序(假设配置了“spring.integration.rsocket.server.message-mapping-enabled”)。

Spring Boot还可以使用配置属性自动配置客户端RSocketConnector：

Properties

Yaml

# 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

Properties

Yaml

# 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"

有关更多详细信息，请参阅IntegrationAutoConfiguration和IntegrationProperties类。

6. WebSockets

Spring Boot为嵌入式Tomcat、Jetty和Undertow提供WebSockets自动配置。如果将WAR文件部署到独立容器，则Spring Boot假定该容器负责其WebSocket支持的配置。

Spring框架为MVC Web应用程序提供了丰富的WebSocket支持，这些应用程序可以通过Spring-ot-starter-web套接字模块轻松访问。

WebSocket支持也可用于反应性Web应用程序，并且需要将WebSocket API与Spring-ot-starter-webflow一起包括：

<dependency>
    <groupId>javax.websocket</groupId>
    <artifactId>javax.websocket-api</artifactId>
</dependency>

7. What to Read Next

下一节介绍如何在应用程序中启用IO功能。您可以在本节中阅读有关缓存、邮件、验证、REST客户端等内容。