英文原文https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-messaging.html目录JMSActiveMQ支持Artemis支持使用JNDI ConnectionFactory发送消息接收消息AMQPRabbitMQ支持发送消息接收消息Apache Kafka支持发送消息接收消息Kafka流其他Kafka PropertiesSpring Framework为与消息传递系统的集成提供了广泛的支持从使用JmsTemplate简化JMS API的使用到异步接收消息的完整基础结构。 Spring AMQP为高级消息队列协议提供了类似的功能集。 Spring Boot还为RabbitTemplate和RabbitMQ提供自动配置选项。 Spring WebSocket本身包含对STOMP消息传递的支持Spring Boot通过启动器和少量自动配置支持它。 Spring Boot也支持Apache Kafka。JMSjavax.jms.ConnectionFactory接口提供了一种创建javax.jms.Connection以与JMS代理进行交互的标准方法。 虽然Spring需要一个ConnectionFactory来与JMS一起工作但是你通常不需要自己直接使用它而是可以依赖更高级别的消息传递抽象。 有关详细信息请参阅Spring Framework参考文档的相关部分。Spring Boot还会自动配置发送和接收消息所需的基础结构。ActiveMQ支持当ActiveMQ在类路径上可用时Spring Boot也可以配置ConnectionFactory。 如果代理存在则会自动启动并配置嵌入式代理前提是未通过配置指定代理URL。如果使用spring-boot-starter-activemq则提供连接或嵌入ActiveMQ实例的必要依赖项以及与JMS集成的Spring基础结构。ActiveMQ配置由spring.activemq.*中的外部配置属性控制。 例如您可以在application.properties中声明以下部分spring.activemq.broker-urltcp://192.168.1.210:9876 spring.activemq.useradmin spring.activemq.passwordsecret默认情况下CachingConnectionFactory将本机ConnectionFactory包装为可由spring.jms.*中的外部配置属性控制的合理设置:spring.jms.cache.session-cache-size5如果您更愿意使用本机池则可以通过向org.messaginghub:pooled-jms添加依赖项并相应地配置JmsPoolConnectionFactory来实现如以下示例所示spring.activemq.pool.enabledtrue spring.activemq.pool.max-connections50有关更多支持的选项请参阅ActiveMQProperties。 您还可以注册实现ActiveMQConnectionFactoryCustomizer的任意数量的bean以进行更高级的自定义。默认情况下ActiveMQ会创建一个目标如果它尚不存在以便根据提供的名称解析目标。Artemis支持Spring Boot可以在检测到Artemis在类路径上可用时自动配置ConnectionFactory。 如果存在代理则会自动启动并配置嵌入式代理除非已明确设置mode属性。 嵌入支持的模式明确表示需要嵌入式代理如果代理在类路径上不可用则发生错误和本机使用netty传输协议连接到代理。 配置后者后Spring Boot会使用默认设置配置ConnectionFactory该ConnectionFactory连接到在本地计算机上运行的代理。如果使用spring-boot-starter-artemis则会提供连接到现有Artemis实例的必要依赖项以及与JMS集成的Spring基础结构。 将org.apache.activemq:artemis-jms-server添加到您的应用程序可让您使用嵌入模式。Artemis配置由spring.artemis.*中的外部配置属性控制。 例如您可以在application.properties中声明以下部分spring.artemis.modenative spring.artemis.host192.168.1.210 spring.artemis.port9876 spring.artemis.useradmin spring.artemis.passwordsecret嵌入代理时您可以选择是否要启用持久性并列出应该可用的目标。 可以将这些指定为以逗号分隔的列表以使用默认选项创建它们也可以定义类型为org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration或org.apache.activemq的bean。 artemis.jms.server.config.TopicConfiguration分别用于高级队列和主题配置。默认情况下CachingConnectionFactory将本机ConnectionFactory包装为可由spring.jms.*中的外部配置属性控制的合理设置spring.jms.cache.session-cache-size5如果您更愿意使用本机池则可以通过向org.messaginghub:pooled-jms添加依赖项并相应地配置JmsPoolConnectionFactory来实现如以下示例所示spring.artemis.pool.enabledtrue spring.artemis.pool.max-connections50有关更多支持的选项请参阅ArtemisProperties。不涉及JNDI查找并使用Artemis配置中的name属性或通过配置提供的名称来解析目标名称。使用JNDI ConnectionFactory如果您在应用程序服务器中运行应用程序Spring Boot会尝试使用JNDI找到JMS ConnectionFactory。 默认情况下将检查java:/JmsXA和java:/XAConnectionFactory位置。 如果需要指定备用位置可以使用spring.jms.jndi-name属性如以下示例所示spring.jms.jndi-namejava:/MyConnectionFactory发送消息Spring的JmsTemplate是自动配置的你可以直接将它自动装入你自己的bean中如下例所示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可以以类似的方式注入。 如果定义了DestinationResolver或MessageConverter bean它将自动关联到自动配置的JmsTemplate。接收消息当存在JMS基础结构时可以使用JmsListener对任何bean进行批注以创建侦听器端点。如果未定义JmsListenerContainerFactory则会自动配置默认值。如果定义了DestinationResolver或MessageConverter bean它将自动关联到默认工厂。默认情况下默认工厂是事务性的。如果您在存在JtaTransactionManager的基础结构中运行则默认情况下它与侦听器容器相关联。如果不是则启用sessionTransacted标志。在后一种情况下您可以通过在侦听器方法或其委托上添加Transactional将本地数据存储事务与传入消息的处理相关联。这确保了在本地事务完成后确认传入消息。这还包括发送已在同一JMS会话上执行的响应消息。以下组件在someQueue目标上创建一个侦听器端点Component public class MyBean { JmsListener(destination someQueue) public void processMessage(String content) { // ... } }有关更多详细信息请参阅EnableJms的Javadoc。如果需要创建更多JmsListenerContainerFactory实例或者如果要覆盖缺省值Spring Boot会提供一个DefaultJmsListenerContainerFactoryConfigurer您可以使用它来初始化DefaultJmsListenerContainerFactory其设置与自动配置的设置相同。例如以下示例公开了另一个使用特定MessageConverter的工厂Configuration static class JmsConfiguration { Bean public DefaultJmsListenerContainerFactory myFactory( DefaultJmsListenerContainerFactoryConfigurer configurer) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); configurer.configure(factory, connectionFactory()); factory.setMessageConverter(myMessageConverter()); return factory; } }然后您可以在任何JmsListener-annotated方法中使用工厂如下所示Component public class MyBean { JmsListener(destination someQueue, containerFactorymyFactory) public void processMessage(String content) { // ... } }AMQP高级消息队列协议AMQP是面向消息的中间件的平台中立的线级协议。 Spring AMQP项目将核心Spring概念应用于基于AMQP的消息传递解决方案的开发。 Spring Boot为通过RabbitMQ使用AMQP提供了一些便利包括spring-boot-starter-amqp“Starter”。RabbitMQ支持RabbitMQ是一个基于AMQP协议的轻量级可靠可扩展且可移植的消息代理。 Spring使用RabbitMQ通过AMQP协议进行通信。RabbitMQ配置由spring.rabbitmq.*中的外部配置属性控制。 例如您可以在application.properties中声明以下部分spring.rabbitmq.hostlocalhost spring.rabbitmq.port5672 spring.rabbitmq.usernameadmin spring.rabbitmq.passwordsecret如果上下文中存在ConnectionNameStrategy bean它将自动用于命名由自动配置的ConnectionFactory创建的连接。 有关更多支持的选项请参阅RabbitProperties。有关详细信息请参阅了解AMQPRabbitMQ使用的协议。发送消息Spring的AmqpTemplate和AmqpAdmin是自动配置的您可以将它们直接自动装入自己的bean中如下例所示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可以以类似的方式注入。 如果定义了MessageConverter bean它将自动关联到自动配置的AmqpTemplate。如有必要任何定义为bean的org.springframework.amqp.core.Queue都会自动用于在RabbitMQ实例上声明相应的队列。要重试操作可以在AmqpTemplate上启用重试例如在代理连接丢失的情况下spring.rabbitmq.template.retry.enabledtrue spring.rabbitmq.template.retry.initial-interval2s默认情况下禁用重试。 您还可以通过声明RabbitRetryTemplateCustomizer bean以编程方式自定义RetryTemplate。接收消息当Rabbit基础结构存在时可以使用RabbitListener注释任何bean以创建侦听器端点。 如果未定义RabbitListenerContainerFactory则会自动配置默认的SimpleRabbitListenerContainerFactory您可以使用spring.rabbitmq.listener.type属性切换到直接容器。 如果定义了MessageConverter或MessageRecoverer bean它将自动与默认工厂关联。以下示例组件在someQueue队列上创建一个侦听器端点Component public class MyBean { RabbitListener(queues someQueue) public void processMessage(String content) { // ... } }有关更多详细信息请参阅EnableRabbit的Javadoc。如果需要创建更多RabbitListenerContainerFactory实例或者如果要覆盖缺省值Spring Boot提供了一个SimpleRabbitListenerContainerFactoryConfigurer和一个DirectRabbitListenerContainerFactoryConfigurer您可以使用它来初始化SimpleRabbitListenerContainerFactory和DirectRabbitListenerContainerFactory其设置与自动配置使用的工厂相同。您选择的容器类型无关紧要。 这两个bean通过自动配置公开。例如以下配置类公开了另一个使用特定MessageConverter的工厂Configuration static class RabbitConfiguration { Bean public SimpleRabbitListenerContainerFactory myFactory( SimpleRabbitListenerContainerFactoryConfigurer configurer) { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); configurer.configure(factory, connectionFactory); factory.setMessageConverter(myMessageConverter()); return factory; } }然后您可以在任何RabbitListener-annotated方法中使用工厂如下所示Component public class MyBean { RabbitListener(queues someQueue, containerFactorymyFactory) public void processMessage(String content) { // ... } }您可以启用重试来处理侦听器抛出异常的情况。 默认情况下使用RejectAndDontRequeueRecoverer但您可以定义自己的MessageRecoverer。 当重试耗尽时如果代理配置了这样做则拒绝该消息并将其丢弃或路由到死信交换。 默认情况下禁用重试。 您还可以通过声明RabbitRetryTemplateCustomizer bean以编程方式自定义RetryTemplate。重要默认情况下如果禁用重试并且侦听器抛出异常则会无限期地重试传递。 您可以通过两种方式修改此行为将defaultRequeueRejected属性设置为false以便尝试零重新传递或抛出AmqpRejectAndDontRequeueException以指示应拒绝该消息。 后者是启用重试并且达到最大传递尝试次数时使用的机制。Apache Kafka支持通过提供spring-kafka项目的自动配置来支持Apache Kafka。Kafka配置由spring.kafka.*中的外部配置属性控制。 例如您可以在application.properties中声明以下部分spring.kafka.bootstrap-serverslocalhost:9092 spring.kafka.consumer.group-idmyGroup要在启动时创建主题请添加NewTopic类型的Bean。 如果主题已存在则忽略该bean。有关更多支持的选项请参阅KafkaProperties。发送消息Spring的KafkaTemplate是自动配置的您可以直接在自己的bean中自动装配它如下例所示Component public class MyBean { private final KafkaTemplate kafkaTemplate; Autowired public MyBean(KafkaTemplate kafkaTemplate) { this.kafkaTemplate kafkaTemplate; } // ... }如果定义了属性spring.kafka.producer.transaction-id-prefix则会自动配置KafkaTransactionManager。 此外如果定义了RecordMessageConverter bean它将自动关联到自动配置的KafkaTemplate。接收消息当存在Apache Kafka基础结构时可以使用KafkaListener注释任何bean以创建侦听器端点。 如果未定义KafkaListenerContainerFactory则会使用spring.kafka.listener.*中定义的键自动配置默认值。以下组件在someTopic主题上创建一个侦听器端点Component public class MyBean { KafkaListener(topics someTopic) public void processMessage(String content) { // ... } }如果定义了KafkaTransactionManager bean它将自动关联到容器工厂。 同样如果定义了RecordMessageConverterErrorHandler或AfterRollbackProcessor bean它将自动关联到默认工厂。自定义ChainedKafkaTransactionManager必须标记为Primary因为它通常引用自动配置的KafkaTransactionManager bean。Kafka流Spring for Apache Kafka提供了一个工厂bean来创建StreamsBuilder对象并管理其流的生命周期。 只要kafka-streams在类路径上并且通过EnableKafkaStreams注释启用Kafka StreamsSpring Boot就会自动配置所需的KafkaStreamsConfiguration bean。启用Kafka Streams意味着必须设置应用程序ID和引导程序服务器。 可以使用spring.kafka.streams.application-id配置前者如果未设置则默认为spring.application.name。 后者可以全局设置或专门为流而重写。使用专用属性可以使用其他几个属性; 可以使用spring.kafka.streams.properties命名空间设置其他任意Kafka属性。 有关更多信息另请参见第33.3.4节“其他Kafka属性”。要使用工厂bean只需将StreamsBuilder连接到Bean如以下示例所示Configuration EnableKafkaStreams static class KafkaStreamsExampleConfiguration { Bean public KStreamInteger, String kStream(StreamsBuilder streamsBuilder) { KStreamInteger, String stream streamsBuilder.stream(ks1In); stream.map((k, v) - new KeyValue(k, v.toUpperCase())).to(ks1Out, Produced.with(Serdes.Integer(), new JsonSerde())); return stream; } }默认情况下由其创建的StreamBuilder对象管理的流将自动启动。 您可以使用spring.kafka.streams.auto-startup属性自定义此行为。其他Kafka Properties自动配置支持的属性显示在附录A常见应用程序属性中。 请注意在大多数情况下这些属性连字符或camelCase直接映射到Apache Kafka点状属性。 有关详细信息请参阅Apache Kafka文档。这些属性中的前几个适用于所有组件生产者使用者管理员和流但如果您希望使用不同的值则可以在组件级别指定。 Apache Kafka指定重要性为HIGHMEDIUM或LOW的属性。 Spring Boot自动配置支持所有HIGH重要性属性一些选择的MEDIUM和LOW属性以及任何没有默认值的属性。只有Kafka支持的属性的子集可以直接通过KafkaProperties类获得。 如果您希望使用不直接支持的其他属性配置生产者或使用者请使用以下属性spring.kafka.properties.prop.onefirst spring.kafka.admin.properties.prop.twosecond spring.kafka.consumer.properties.prop.threethird spring.kafka.producer.properties.prop.fourfourth spring.kafka.streams.properties.prop.fivefifth这将常见的prop.one Kafka属性设置为第一个适用于生产者消费者和管理员prop.two管理员属性为第二个prop.three消费者属性为第三个prop.four生产者属性为第四个和prop.five streams属性为第五。您还可以按如下方式配置Spring Kafka JsonDeserializerspring.kafka.consumer.value-deserializerorg.springframework.kafka.support.serializer.JsonDeserializer spring.kafka.consumer.properties.spring.json.value.default.typecom.example.Invoice spring.kafka.consumer.properties.spring.json.trusted.packagescom.example,org.acme同样您可以禁用JsonSerializer在标头中发送类型信息的默认行为spring.kafka.producer.value-serializerorg.springframework.kafka.support.serializer.JsonSerializer spring.kafka.producer.properties.spring.json.add.type.headersfalse重要:以这种方式设置的属性将覆盖Spring Boot明确支持的任何配置项。