Spring Boot 消息传递
Spring Framework 为与消息传递系统的集成提供了广泛的支持,从使用 JMS API 的简化使用JmsTemplate
到异步接收消息的完整基础架构。Spring AMQP 为高级消息队列协议提供了类似的功能集。RabbitTemplate
Spring Boot 还为RabbitMQ提供了自动配置选项。Spring WebSocket 原生包括对 STOMP 消息传递的支持,Spring Boot 通过启动器和少量的自动配置来支持它。Spring Boot 还支持 Apache Kafka。
1. JMS
该javax.jms.ConnectionFactory
接口提供了一种创建javax.jms.Connection
用于与 JMS 代理交互的标准方法。尽管 Spring 需要ConnectionFactory
与 JMS 一起使用,但您通常不需要自己直接使用它,而是可以依赖更高级别的消息传递抽象。(有关详细信息,请参阅 Spring Framework 参考文档的相关部分。)Spring Boot 还自动配置必要的基础设施来发送和接收消息。
1.1ActiveMQ 支持
当ActiveMQ在类路径上可用时,Spring Boot 还可以配置一个ConnectionFactory
. 如果存在代理,则会自动启动和配置嵌入式代理(前提是未通过配置指定代理 URL,并且未在配置中禁用嵌入式代理)。
ActiveMQ 配置由spring.activemq.*
.
默认情况下,ActiveMQ 自动配置为使用VM 传输,它启动嵌入在同一 JVM 实例中的代理。
您可以通过配置属性来禁用嵌入式代理,spring.activemq.in-memory
如下例所示:
特性
如果您配置代理 URL,嵌入式代理也将被禁用,如以下示例所示:
特性
如果您想完全控制嵌入式代理,请参阅ActiveMQ 文档以获取更多信息。
默认情况下,a使用您可以通过外部配置属性控制的合理设置CachingConnectionFactory
包装本机:ConnectionFactory
spring.jms.*
特性
如果您更愿意使用本机池,可以通过添加依赖项org.messaginghub:pooled-jms
并相应地配置来实现JmsPoolConnectionFactory
,如以下示例所示:
特性
默认情况下,如果目标尚不存在,ActiveMQ 会创建一个目标,以便根据提供的名称解析目标。
1.2. ActiveMQ Artemis 支持
ConnectionFactory
当 Spring Boot检测到ActiveMQ Artemis在类路径上可用时,它可以自动配置。如果存在代理,则会自动启动和配置嵌入式代理(除非已显式设置模式属性)。支持的模式是embedded
(明确需要嵌入式代理,并且如果代理在类路径上不可用,则应发生错误)和(使用传输协议native
连接到代理)。netty
配置后者时,Spring Boot 配置一个ConnectionFactory
连接到本地机器上运行的代理的默认设置。
ActiveMQ Artemis 配置由spring.artemis.*
. 例如,您可以在 中声明以下部分application.properties
:
特性
嵌入代理时,您可以选择是否要启用持久性并列出应该可用的目标。这些可以指定为逗号分隔的列表以使用默认选项创建它们,或者您可以分别为高级队列和主题配置定义类型为org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration
或的 bean。org.apache.activemq.artemis.jms.server.config.TopicConfiguration
默认情况下,a使用您可以通过外部配置属性控制的合理设置CachingConnectionFactory
包装本机:ConnectionFactory
spring.jms.*
特性
如果您更愿意使用本机池,可以通过添加依赖项org.messaginghub:pooled-jms
并相应地配置来实现JmsPoolConnectionFactory
,如以下示例所示:
特性
有关ArtemisProperties更多支持的选项,请参阅。
不涉及 JNDI 查找,并且使用name
Artemis 配置中的属性或通过配置提供的名称来根据其名称解析目的地。
1.3. 使用 JNDI 连接工厂
如果您在应用程序服务器中运行应用程序,Spring Boot 会尝试ConnectionFactory
使用 JNDI 定位 JMS。默认情况下,检查java:/JmsXA
和java:/XAConnectionFactory
位置。spring.jms.jndi-name
如果需要指定替代位置,可以使用该属性,如以下示例所示:
特性
1.4. 发送消息
SpringJmsTemplate
是自动配置的,您可以直接将其自动装配到您自己的 bean 中,如下例所示:
1.5接收消息
当 JMS 基础设施存在时,可以对任何 bean 进行注释@JmsListener
以创建侦听器端点。如果没有JmsListenerContainerFactory
定义,则自动配置一个默认值。如果定义了 a DestinationResolver
、 aMessageConverter
或 a javax.jms.ExceptionListener
bean,它们将自动与默认工厂关联。
默认情况下,默认工厂是事务性的。如果您在存在 a 的基础架构中运行JtaTransactionManager
,则默认情况下它与侦听器容器相关联。如果不是,sessionTransacted
则启用该标志。在后一种情况下,您可以通过添加@Transactional
侦听器方法(或其委托)将本地数据存储事务与传入消息的处理相关联。这可确保在本地事务完成后确认传入消息。这还包括发送已在同一 JMS 会话上执行的响应消息。
以下组件在someQueue
目标上创建侦听器端点:
如果你需要创建更多JmsListenerContainerFactory
实例或者你想覆盖默认值,Spring Boot 提供了一个DefaultJmsListenerContainerFactoryConfigurer
你可以使用DefaultJmsListenerContainerFactory
与自动配置相同的设置来初始化 a 的实例。
例如,以下示例公开了另一个使用特定的工厂MessageConverter
:
爪哇
科特林
然后您可以在任何带注释的方法中使用工厂,@JmsListener
如下所示:
2. AMQP
高级消息队列协议 (AMQP) 是面向消息中间件的平台中立、线路级协议。Spring AMQP 项目将核心 Spring 概念应用于基于 AMQP 的消息传递解决方案的开发。Spring Boot 为通过 RabbitMQ 使用 AMQP 提供了多种便利,包括spring-boot-starter-amqp
“Starter”。
2.1RabbitMQ 支持
RabbitMQ是基于 AMQP 协议的轻量级、可靠、可扩展、可移植的消息代理。Spring 使用RabbitMQ
通过 AMQP 协议进行通信。
RabbitMQ 配置由spring.rabbitmq.*
. 例如,您可以在 中声明以下部分application.properties
:
特性
或者,您可以使用以下addresses
属性配置相同的连接:
特性
有关RabbitProperties更多受支持的基于属性的配置选项,请参阅。要配置 Spring AMQP 使用的 RabbitMQ 的较低级别的详细信息ConnectionFactory
,请定义一个ConnectionFactoryCustomizer
bean。
如果ConnectionNameStrategy
上下文中存在 bean,它将自动用于命名由 auto-configured 创建的连接CachingConnectionFactory
。
2.2. 发送消息
Spring 的AmqpTemplate
和AmqpAdmin
是自动配置的,您可以将它们直接自动装配到您自己的 bean 中,如下例所示:
如有必要,任何org.springframework.amqp.core.Queue
定义为 bean 的内容都会自动用于在 RabbitMQ 实例上声明相应的队列。
要重试操作,您可以启用重试AmqpTemplate
(例如,在代理连接丢失的情况下):
特性
默认情况下禁用重试。您还可以RetryTemplate
通过声明RabbitRetryTemplateCustomizer
bean 以编程方式自定义。
如果你需要创建更多RabbitTemplate
实例或者你想覆盖默认值,Spring Boot 提供了一个RabbitTemplateConfigurer
bean,你可以使用它来初始化一个RabbitTemplate
与自动配置使用的工厂相同的设置。
2.3. 向流发送消息
要将消息发送到特定流,请指定流的名称,如以下示例所示:
特性
如果定义了MessageConverter
、StreamMessageConverter
或ProducerCustomizer
bean,它会自动关联到 auto-configured RabbitStreamTemplate
。
如果你需要创建更多RabbitStreamTemplate
实例或者你想覆盖默认值,Spring Boot 提供了一个RabbitStreamTemplateConfigurer
bean,你可以使用它来初始化一个RabbitStreamTemplate
与自动配置使用的工厂相同的设置。
2.4. 接收消息
当 Rabbit 基础设施存在时,可以对任何 bean 进行注释@RabbitListener
以创建侦听器端点。如果没有RabbitListenerContainerFactory
定义,则自动配置默认值,您可以使用该属性SimpleRabbitListenerContainerFactory
切换到直接容器。spring.rabbitmq.listener.type
如果定义了 aMessageConverter
或MessageRecoverer
bean,它会自动与默认工厂关联。
以下示例组件在someQueue
队列上创建一个侦听器端点:
如果您需要创建更多RabbitListenerContainerFactory
实例或想要覆盖默认值,Spring Boot 提供了 aSimpleRabbitListenerContainerFactoryConfigurer
和 a DirectRabbitListenerContainerFactoryConfigurer
,您可以使用它们来初始化 aSimpleRabbitListenerContainerFactory
和 a DirectRabbitListenerContainerFactory
,其设置与自动配置使用的工厂相同。
例如,以下配置类公开了另一个使用特定的工厂MessageConverter
:
然后您可以在任何带注释的方法中使用工厂@RabbitListener
,如下所示:
您可以启用重试以处理侦听器引发异常的情况。默认情况下,RejectAndDontRequeueRecoverer
使用,但您可以定义MessageRecoverer
自己的。当重试用尽时,如果代理配置为这样做,则消息将被拒绝并丢弃或路由到死信交换。默认情况下,重试被禁用。您还可以RetryTemplate
通过声明RabbitRetryTemplateCustomizer
bean 以编程方式自定义。
3. Apache Kafka 支持
通过提供项目的自动配置来支持Apache Kafkaspring-kafka
。
Kafka 配置由spring.kafka.*
. 例如,您可以在 中声明以下部分application.properties
:
特性
有关KafkaProperties更多支持的选项,请参阅。
3.1发送消息
SpringKafkaTemplate
是自动配置的,您可以直接在自己的 bean 中自动装配它,如下例所示:
3.2. 接收消息
当存在 Apache Kafka 基础架构时,可以使用任何 bean 进行注释@KafkaListener
以创建侦听器端点。如果没有KafkaListenerContainerFactory
定义,则自动配置一个默认值,其中定义了spring.kafka.listener.*
.
以下组件在someTopic
主题上创建侦听器端点:
如果KafkaTransactionManager
定义了 bean,它会自动关联到容器工厂。类似地,如果定义了RecordFilterStrategy
、或bean CommonErrorHandler
,它会自动关联到默认工厂。AfterRollbackProcessor
ConsumerAwareRebalanceListener
根据侦听器类型,一个RecordMessageConverter
或BatchMessageConverter
bean 与默认工厂相关联。如果RecordMessageConverter
批处理侦听器仅存在一个 bean,则将其包装在BatchMessageConverter
.
3.3. 卡夫卡流
Spring for Apache Kafka 提供了一个工厂 bean 来创建StreamsBuilder
对象并管理其流的生命周期。KafkaStreamsConfiguration
只要在类路径上, Spring Boot 就会自动配置所需的bean,并且注释kafka-streams
启用了 Kafka Streams 。@EnableKafkaStreams
启用 Kafka Streams 意味着必须设置应用程序 ID 和引导服务器。前者可以使用 配置,如果未设置则spring.kafka.streams.application-id
默认为。spring.application.name
后者可以全局设置或仅针对流专门覆盖。
使用专用属性可以使用几个附加属性;可以使用spring.kafka.streams.properties
命名空间设置其他任意 Kafka 属性。另请参阅其他 Kafka 属性以获取更多信息。
要使用工厂 bean,请按照以下示例所示连接StreamsBuilder
到您的 bean:@Bean
默认情况下,由StreamBuilder
它创建的对象管理的流是自动启动的。spring.kafka.streams.auto-startup
您可以使用该属性自定义此行为。
3.4. 其他 Kafka 属性
自动配置支持的属性显示在附录的“集成属性”部分。请注意,在大多数情况下,这些属性(连字符或驼峰式)直接映射到 Apache Kafka 点属性。有关详细信息,请参阅 Apache Kafka 文档。
这些属性中的前几个适用于所有组件(生产者、消费者、管理员和流),但如果您希望使用不同的值,可以在组件级别指定。Apache Kafka 指定具有 HIGH、MEDIUM 或 LOW 重要性的属性。Spring Boot 自动配置支持所有 HIGH 重要性属性、一些选定的 MEDIUM 和 LOW 属性以及任何没有默认值的属性。
只有 Kafka 支持的属性的一个子集可以直接通过KafkaProperties
该类获得。如果您希望使用不直接支持的其他属性配置生产者或消费者,请使用以下属性:
特性
这将通用prop.one
Kafka 属性设置为first
(适用于生产者、消费者和管理员),将prop.two
admin 属性设置为second
,将prop.three
消费者属性设置为third
,将prop.four
生产者属性设置为fourth
,将prop.five
流属性设置为fifth
。
您还可以JsonDeserializer
按如下方式配置 Spring Kafka:
特性
同样,您可以禁用JsonSerializer
在标头中发送类型信息的默认行为:
特性
3.5. 使用嵌入式 Kafka 进行测试
Spring for Apache Kafka 提供了一种使用嵌入式 Apache Kafka 代理测试项目的便捷方式。要使用此功能,请使用模块注释测试@EmbeddedKafka
类spring-kafka-test
。有关更多信息,请参阅 Spring for Apache Kafka参考手册。
要使 Spring Boot 自动配置与上述嵌入式 Apache Kafka 代理一起工作,您需要将嵌入式代理地址(由 填充EmbeddedKafkaBroker
)的系统属性重新映射到 Apache Kafka 的 Spring Boot 配置属性中。有几种方法可以做到这一点:
- 提供一个系统属性以将嵌入的代理地址映射到
spring.kafka.bootstrap-servers
测试类中:
-
@EmbeddedKafka
在注解上配置一个属性名称:
- 在配置属性中使用占位符:
特性
4.RSocket
RSocket是用于字节流传输的二进制协议。它通过通过单个连接传递的异步消息启用对称交互模型。
Spring Framework的spring-messaging
模块为客户端和服务器端的 RSocket 请求者和响应者提供支持。有关更多详细信息,请参阅 Spring Framework 参考的RSocket 部分,包括 RSocket 协议的概述。
4.1RSocket 策略自动配置
Spring Boot 自动配置一个RSocketStrategies
bean,该 bean 提供编码和解码 RSocket 有效负载所需的所有基础设施。默认情况下,自动配置将尝试配置以下内容(按顺序):
- Jackson 的CBOR编解码器
- 杰克逊的 JSON 编解码器
启动器提供了这spring-boot-starter-rsocket
两个依赖项。请参阅Jackson 支持部分以了解有关自定义可能性的更多信息。
开发人员可以RSocketStrategies
通过创建实现RSocketStrategiesCustomizer
接口的 bean 来自定义组件。请注意,它们@Order
很重要,因为它决定了编解码器的顺序。
4.2. RSocket 服务器自动配置
Spring Boot 提供 RSocket 服务器自动配置。所需的依赖项由spring-boot-starter-rsocket
.
Spring Boot 允许从 WebFlux 服务器通过 WebSocket 公开 RSocket,或者建立一个独立的 RSocket 服务器。这取决于应用程序的类型及其配置。
对于 WebFlux 应用程序(类型为WebApplicationType.REACTIVE
),仅当以下属性匹配时,RSocket 服务器才会插入 Web 服务器:
特性
或者,RSocket TCP 或 websocket 服务器作为独立的嵌入式服务器启动。除了依赖项要求之外,唯一需要的配置是为该服务器定义一个端口:
特性
4.3. Spring 消息传递 RSocket 支持
Spring Boot 将为 RSocket 自动配置 Spring Messaging 基础设施。
这意味着 Spring Boot 将创建一个RSocketMessageHandler
bean 来处理对应用程序的 RSocket 请求。
4.4. 使用 RSocketRequester 调用 RSocket 服务
一旦RSocket
服务器和客户端之间建立了通道,任何一方都可以向另一方发送或接收请求。
作为服务器,您可以RSocketRequester
在 RSocket 的任何处理程序方法上注入实例@Controller
。作为客户端,首先需要配置并建立 RSocket 连接。RSocketRequester.Builder
Spring Boot使用预期的编解码器为这种情况自动配置一个并应用任何RSocketConnectorConfigurer
bean。
该RSocketRequester.Builder
实例是一个原型 bean,这意味着每个注入点都会为您提供一个新的实例。这是故意这样做的,因为此构建器是有状态的,您不应使用同一实例创建具有不同设置的请求者。
下面的代码展示了一个典型的例子:
5. 弹簧集成
Spring Boot 为使用Spring Integration提供了多种便利,包括spring-boot-starter-integration
“Starter”。Spring Integration 提供了对消息传递以及 HTTP、TCP 等其他传输的抽象。如果 Spring Integration 在您的类路径上可用,它会通过@EnableIntegration
注解进行初始化。
Spring Integration 轮询逻辑依赖于自动配置的TaskScheduler. 可以使用配置属性自定义默认值PollerMetadata
(每秒轮询无限数量的消息) 。spring.integration.poller.*
Spring Boot 还配置了一些由额外的 Spring Integration 模块触发的特性。如果spring-integration-jmx
也在类路径上,则通过 JMX 发布消息处理统计信息。如果spring-integration-jdbc
可用,则可以在启动时创建默认数据库模式,如下行所示:
特性
如果spring-integration-rsocket
可用,开发人员可以使用"spring.rsocket.server.*"
属性配置 RSocket 服务器,并让它使用IntegrationRSocketEndpoint
或RSocketOutboundGateway
组件来处理传入的 RSocket 消息。此基础设施可以处理 Spring Integration RSocket 通道适配器和@MessageMapping
处理程序("spring.integration.rsocket.server.message-mapping-enabled"
已配置)。
Spring Boot 还可以自动配置ClientRSocketConnector
using 配置属性:
特性
特性
有关更多详细信息,请参阅IntegrationAutoConfiguration和IntegrationProperties类。
6. WebSockets
Spring Boot 为嵌入式 Tomcat、Jetty 和 Undertow 提供 WebSockets 自动配置。如果将 war 文件部署到独立容器,Spring Boot 假定容器负责配置其 WebSocket 支持。
Spring Framework 为 MVC Web 应用程序提供了丰富的 WebSocket 支持,可以通过spring-boot-starter-websocket
模块轻松访问。
WebSocket 支持也可用于响应式 Web 应用程序,并且需要同时包含 WebSocket API spring-boot-starter-webflux
: