Spring Boot Messaging中文文档

本文为官方文档直译版本。原文链接

引言

Spring 框架为与消息系统集成提供了广泛的支持，从使用 JmsTemplate 简化 JMS API 的使用到异步接收消息的完整基础架构。Spring AMQP 为高级消息队列协议提供了类似的功能集。Spring Boot 还为 RabbitTemplate 和 RabbitMQ 提供了自动配置选项。Spring WebSocket 原生包含对 STOMP 消息传递的支持，Spring Boot 通过启动器和少量自动配置对此提供了支持。Spring Boot 还支持 Apache Kafka 和 Apache Pulsar。

JMS

jakarta.jms.ConnectionFactory 接口提供了创建 jakarta.jms.Connection 的标准方法，用于与 JMS 代理交互。虽然 Spring 需要 ConnectionFactory 才能与 JMS 协作，但您一般不需要直接使用它，而是可以依赖更高级别的消息传递抽象。(详情请参见 Spring Framework 参考文档的相关部分）。Spring Boot 还会自动配置发送和接收消息所需的基础架构。

ActiveMQ “Classic” 支持

当 ActiveMQ “Classic” 在类路径上可用时，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"
    user: "admin"
    password: "secret"

默认情况下，缓存连接工厂（CachingConnectionFactory）将本地连接工厂（ConnectionFactory）与合理的设置进行封装，您可以通过 spring.jms.* 中的外部配置属性来控制这些设置：

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

如果你更愿意使用原生池，可以通过为 org.messaginghub:pooled-jms 添加依赖关系并配置相应 JmsPoolConnectionFactory 来实现，如下例所示：

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

请参阅 ActiveMQProperties 了解更多支持的选项。您还可以注册任意数量的实现 ActiveMQConnectionFactoryCustomizer 的 Bean，以进行更高级的定制。

默认情况下，如果目的地（destinations）不存在，ActiveMQ “Classic” 会创建该目的地，以便根据提供的名称解析目的地。

ActiveMQ Artemis支持

当 Spring Boot 检测到类路径上有 ActiveMQ Artemis 时，它可以自动配置 ConnectionFactory。如果存在代理，则会自动启动和配置嵌入式代理（除非显式设置了模式属性）。支持的模式有嵌入式（明确表示需要嵌入式代理，如果代理在类路径上不可用，则会出错）和原生（使用 netty 传输协议连接代理）。当配置后者时，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
    broker-url: "tcp://192.168.1.210:9876"
    user: "admin"
    password: "secret"

嵌入代理时，您可以选择是否要启用持久性，并列出应提供的目的地（destinations）。您可以以逗号分隔列表的形式指定这些选项，以便使用默认选项创建它们，也可以定义 org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration 或 org.apache.activemq.artemis.jms.server.config.TopicConfiguration 类型的 bean，分别用于高级队列和主题配置。
默认情况下，缓存连接工厂（CachingConnectionFactory）将原生连接工厂（Native ConnectionFactory）与可通过 spring.jms.* 中的外部配置属性控制的合理设置封装在一起：

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

如果你想使用原生池，可以添加 org.messaginghub:pooled-jms 的依赖，并相应配置 JmsPoolConnectionFactory，如下例所示：

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

更多支持的选项请参见 ArtemisProperties。
不涉及 JNDI 查找，目的地（destinations）将使用 ActiveMQ Artemis 配置中的name属性或通过配置提供的名称，根据其名称进行解析。

使用 JNDI ConnectionFactory

如果在应用服务器中运行应用程序，Spring Boot 会尝试使用 JNDI 查找 JMS ConnectionFactory。默认情况下，会检查 java:/JmsXA 和 java:/XAConnectionFactory 位置。如果需要指定其他位置，可以使用 spring.jms.jndi-name 属性，如下例所示：

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

发送消息

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

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");
    }

}

JmsMessagingTemplate 可以使用类似的方式注入。如果定义了 DestinationResolver 或 MessageConverter Bean，它会自动与自动配置的 JmsTemplate 关联。

接收消息

当 JMS 基础设施存在时，任何 bean 都可以用 @JmsListener 进行注解，以创建一个监听器端点。如果没有定义 JmsListenerContainerFactory，则会自动配置一个默认的 JmsListenerContainerFactory。如果定义了 DestinationResolver、MessageConverter 或 jakarta.jms.ExceptionListener Bean，它们会自动与默认工厂关联。
默认情况下，默认工厂是事务性的。如果在存在 JtaTransactionManager 的基础架构中运行，默认情况下会将其关联到监听器容器。如果没有，则会启用 sessionTransacted 标志。在后一种情况下，您可以通过在监听器方法（或其委托）上添加 @Transactional 来将本地数据存储事务与传入消息的处理关联起来。这样就能确保本地事务完成后，传入消息得到确认。这也包括发送在同一 JMS 会话中执行的响应消息。
下面的组件在 someQueue 目标上创建了一个监听器端点：

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @JmsListener(destination = "someQueue")
    public void processMessage(String content) {
        // ...
    }

}

更多详情，请参阅 @EnableJms 的 Javadoc

如果您需要创建更多 JmsListenerContainerFactory 实例，或者想覆盖默认值，Spring Boot 提供了一个 DefaultJmsListenerContainerFactoryConfigurer，您可以使用它来初始化一个 DefaultJmsListenerContainerFactory，其设置与自动配置的相同。
例如，下面的示例暴露了另一个使用特定 MessageConverter 的工厂：

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 ...
    }

}

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

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) {
        // ...
    }

}

AMQP

高级消息队列协议（Advanced Message Queuing Protocol，AMQP）是面向消息的中间件的平台中立的线级协议。Spring AMQP 项目将 Spring 核心概念应用于开发基于 AMQP 的消息传递解决方案。Spring Boot 通过 RabbitMQ（包括 spring-boot-starter-amqp “Starter”）为使用 AMQP 提供了多种便利。

RabbitMQ 支持

RabbitMQ 是基于 AMQP 协议的轻量级、可靠、可扩展和可移植的消息代理。Spring 使用 RabbitMQ 通过 AMQP 协议进行通信。
RabbitMQ 配置由 spring.rabbitmq.* 中的外部配置属性控制。例如，您可以在 application.properties 中声明以下部分：

spring:
  rabbitmq:
    host: "localhost"
    port: 5672
    username: "admin"
    password: "secret"

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

spring:
  rabbitmq:
    addresses: "amqp://admin:secret@localhost"

以这种方式指定地址时，主机和端口属性将被忽略。如果地址使用 amqps 协议，则会自动启用 SSL 支持。

请参阅 RabbitProperties 了解更多支持的基于属性的配置选项。要配置 Spring AMQP 使用的 RabbitMQ ConnectionFactory 的底层细节，请定义 ConnectionFactoryCustomizer Bean。
如果上下文中存在 ConnectionNameStrategy Bean，它将自动用于命名由CachingConnectionFactory 自动配置的创建的连接。
要对 RabbitTemplate 进行应用程序范围内的附加自定义，请使用 RabbitTemplateCustomizer Bean。

有关详细信息，请参阅了解 RabbitMQ 使用的协议 AMQP。

发送消息

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

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");
    }

}

RabbitMessagingTemplate 可以使用类似的方式注入。如果定义了 MessageConverter Bean，它将自动与自动配置的 AmqpTemplate 关联。

如有必要，任何定义为 bean 的 org.springframework.amqp.core.Queue 都会自动用于在 RabbitMQ 实例上声明相应的队列。
要重试操作，您可以在 AmqpTemplate 上启用重试（例如，在代理连接丢失的情况下）：

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

重试默认为禁用。您还可以通过声明 RabbitRetryTemplateCustomizer Bean 以编程方式自定义 RetryTemplate。
如果您需要创建更多 RabbitTemplate 实例或想覆盖默认值，Spring Boot 提供了 RabbitTemplateConfigurer Bean，您可以使用该 Bean 以与自动配置所使用的工厂相同的设置初始化 RabbitTemplate。

向流发送消息

要向特定流发送信息，请指定该流的名称，如下例所示：

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

如果定义了 MessageConverter、StreamMessageConverter 或 ProducerCustomizer Bean，它将自动与自动配置的 RabbitStreamTemplate 关联。
如果您需要创建更多 RabbitStreamTemplate 实例或想覆盖默认值，Spring Boot 提供了 RabbitStreamTemplateConfigurer Bean，您可以使用它来初始化 RabbitStreamTemplate，其设置与自动配置使用的工厂相同。

接收消息

当 Rabbit 基础架构存在时，任何 bean 都可以用 @RabbitListener 进行注解，以创建一个监听器端点。如果没有定义 RabbitListenerContainerFactory，则会自动配置默认的 SimpleRabbitListenerContainerFactory，您可以使用 spring.rabbitmq.listener.type 属性切换到直连容器。如果定义了 MessageConverter 或 MessageRecoverer Bean，它将自动与默认工厂关联。
以下示例组件在 someQueue 队列上创建了一个监听器端点：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @RabbitListener(queues = "someQueue")
    public void processMessage(String content) {
        // ...
    }

}

有关详细信息，请参阅 @EnableRabbit 的 Javadoc

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

选择哪种容器类型并不重要。自动配置会暴露这两个 Bean。

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

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 ...
    }

}

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

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) {
        // ...
    }

}

您可以启用重试来处理监听器抛出异常的情况。默认情况下使用 RejectAndDontRequeueRecoverer，但您也可以定义自己的 MessageRecoverer。当重试次数用尽时，消息将被拒绝，并被丢弃或路由到死信交换（如果代理被配置为这样做）。默认情况下，重试是禁用的。您还可以通过声明 RabbitRetryTemplateCustomizer Bean 以编程方式自定义 RetryTemplate。

默认情况下，如果禁用重试且监听器抛出异常，则会无限期重试传送。您可以通过两种方式修改这种行为：将 defaultRequeueRejected 属性设置为 false，这样尝试的重发次数就为零，或者抛出 AmqpRejectAndDontRequeueException 异常，提示信息应被拒绝。后者是启用重试且达到最大发送尝试次数时使用的机制。

Apache Kafka 支持

通过提供 spring-kafka 项目的自动配置来支持 Apache Kafka。
Kafka 配置由 spring.kafka.* 中的外部配置属性控制。例如，你可以在 application.properties 中声明以下部分：

spring:
  kafka:
    bootstrap-servers: "localhost:9092"
    consumer:
      group-id: "myGroup"

要在启动时创建一个主题，请添加一个 NewTopic 类型的 Bean。如果主题已经存在，该 Bean 将被忽略。

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

发送消息

Spring 的 KafkaTemplate 是自动配置的，你可以直接在自己的 Bean 中注入它，如下例所示：

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");
    }

}

如果定义了属性 spring.kafka.producer.transaction-id-prefix，就会自动配置 KafkaTransactionManager。此外，如果定义了 RecordMessageConverter Bean，它将自动与自动配置的 KafkaTemplate 关联。

接收消息

当 Apache Kafka 基础架构存在时，任何 bean 都可以用 @KafkaListener 进行注解，以创建一个监听器端点。如果没有定义 KafkaListenerContainerFactory，则会自动配置一个默认的 KafkaListenerContainerFactory，其键定义在 spring.kafka.listener.*中。
下面的组件在 someTopic 主题上创建了一个监听器端点：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @KafkaListener(topics = "someTopic")
    public void processMessage(String content) {
        // ...
    }

}

如果定义了 KafkaTransactionManager Bean，它将自动与容器工厂相关联。同样，如果定义了 RecordFilterStrategy、CommonErrorHandler、AfterRollbackProcessor 或 ConsumerAwareRebalanceListener Bean，它将自动关联到默认工厂。
根据监听器类型的不同，记录信息转换器（RecordMessageConverter）或批处理信息转换器（BatchMessageConverter）Bean 会关联到默认工厂。如果批量监听器只有一个 RecordMessageConverter Bean，它将被封装在 BatchMessageConverter 中。

自定义 ChainedKafkaTransactionManager 必须标记为 @Primary，因为它通常会引用自动配置的 KafkaTransactionManager Bean。

Kafka Streams

Spring for Apache Kafka 提供了一个工厂 Bean，用于创建 StreamsBuilder 对象并管理其流的生命周期。只要类路径中包含 kafka-streams，且 Kafka Streams 已启用 @EnableKafkaStreams 注解，Spring Boot 就会自动配置所需的 KafkaStreamsConfiguration Bean。
启用 Kafka Streams 意味着必须设置应用程序 ID 和引导服务器。前者可以使用 spring.kafka.streams.application-id 进行配置，如果未设置，则默认为 spring.application.name。后者可以全局设置，也可以只针对流设置。
使用专用属性还可获得其他一些属性；使用 spring.kafka.streams.properties 命名空间可设置其他任意 Kafka 属性。更多信息，另请参阅“其他的 Kafka 属性”。
要使用工厂 Bean，请将 StreamsBuilder 注入 @Bean 中，如下例所示：

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());
    }

}

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

其他的 Kafka 属性

附录的 “集成属性” 部分显示了自动配置支持的属性。请注意，在大多数情况下，这些属性（hyphenated 或 camelCase）直接映射到 Apache Kafka 的点属性。详情请参见 Apache Kafka 文档。
名称中不包含客户端类型（producer, consumer, admin, 或 streams）的属性被视为通用属性，适用于所有客户端。如果需要，这些通用属性中的大多数都可以针对一种或多种客户端类型进行重载。
Apache Kafka 将属性的重要性分为 “高”（HIGH）、“中”（MEDIUM）或 “低”（LOW）。Spring Boot 自动配置支持所有重要性高的属性、部分选定的重要性中等和低的属性，以及任何没有默认值的属性。
Kafka 支持的属性中只有一部分可直接通过 KafkaProperties 类使用。如果您希望使用不直接支持的其他属性配置各个客户端类型，请使用以下属性：

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"

这将把常用的Kafka 属性 prop.one 设置为first（适用于生产者、消费者、管理员和流），把管理员属性 prop.two 设置为second，把消费者属性 prop.three 设置为third，把生产者属性 prop.four 设置为fourth，把流属性 prop.five 设置为fifth。
你还可以配置 Spring Kafka JsonDeserializer，如下所示：

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"

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

spring:
  kafka:
    producer:
      value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"
      properties:
        "[spring.json.add.type.headers]": false

以这种方式设置的属性会覆盖 Spring Boot 明确支持的任何配置项。

使用嵌入式 Kafka 进行测试

Spring for Apache Kafka 为使用嵌入式 Apache Kafka 代理测试项目提供了一种便捷的方法。要使用这一功能，请使用 spring-kafka-test 模块中的 @EmbeddedKafka 对测试类进行注解。更多信息，请参阅 Spring for Apache Kafka 参考手册。
要使 Spring Boot 自动配置与上述嵌入式 Apache Kafka 代理协同工作，你需要将嵌入式代理地址的系统属性（由 EmbeddedKafkaBroker 填充）重新映射到 Apache Kafka 的 Spring Boot 配置属性中。有几种方法可以做到这一点：

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

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

配置 @EmbeddedKafka 注解的属性名称：

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 {

    // ...

}

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

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

Apache Pulsar 支持

通过提供 Spring for Apache Pulsar 项目的自动配置来支持 Apache Pulsar。
当 org.springframework.pulsar:spring-pulsar 位于类路径上时，Spring Boot 将自动配置和注册 Spring for Apache Pulsar classic（imperative）组件。当类路径上有org.springframework.pulsar:spring-pulsar-reactive时，它也会为反应式组件做同样的工作。
还有 spring-boot-starter-pulsar 和 spring-boot-starter-pulsar-reactive 两种 “Starter”，分别用于方便地收集命令式和反应式使用的依赖关系。

连接 Pulsar

使用 Pulsar Starter时，Spring Boot 将自动配置和注册 PulsarClient Bean。
默认情况下，应用程序会尝试连接到本地 Pulsar 实例，地址为 pulsar://localhost:6650。可以通过将 spring.pulsar.client.service-url 属性设置为不同的值来进行调整。

值必须是有效的 Pulsar 协议 URL

你可以通过指定任何以 spring.pulsar.client.* 为前缀的应用程序属性来配置客户端。
如果需要对配置进行更多控制，可考虑注册一个或多个 PulsarClientBuilderCustomizer Bean。

Authentication

要连接到需要身份验证的 Pulsar 集群，需要通过设置插件类名（pluginClassName）和插件所需的任何参数来指定要使用的身份验证插件。您可以将参数设置为参数名称到参数值的映射。下面的示例展示了如何配置 AuthenticationOAuth2 插件。

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

您需要确保在 spring.pulsar.client.authentication.param.* 下定义的名称与您的 auth 插件所期望的名称完全一致（通常是驼峰格式）。Spring Boot 不会尝试对这些条目进行任何形式的宽松绑定。
例如，如果要为 AuthenticationOAuth2 验证插件配置发行者 URL，就必须使用 spring.pulsar.client.authentication.param.issuerUrl。如果使用 issuerurl 或 issuer-url 等其他形式，设置将不会应用到插件。

SSL

默认情况下，Pulsar 客户端以纯文本与 Pulsar 服务通信。你可以按照 Spring for Apache Pulsar 参考文档中的步骤启用 TLS 加密。
有关客户端和身份验证的完整细节，请参阅 Spring for Apache Pulsar 参考文档。

反应式连接 Pulsar

激活 Reactive 自动配置后，Spring Boot 将自动配置并注册一个 ReactivePulsarClient Bean。
ReactivePulsarClient 将适配之前描述的 PulsarClient 实例。因此，请按照前面的章节配置 ReactivePulsarClient 使用的 PulsarClient。

连接 Pulsar 管理器

Spring for Apache Pulsar 的 PulsarAdministration 客户端也是自动配置的。
默认情况下，应用程序会尝试连接到本地 Pulsar 实例，网址是 http://localhost:8080。可以通过将 spring.pulsar.admin.service-url 属性设置为 (http|https)://<host>:<port> 形式的不同值来进行调整。
如果需要对配置进行更多控制，可考虑注册一个或多个 PulsarAdminBuilderCustomizer Bean。

Authentication

访问需要身份验证的 Pulsar 集群时，管理员客户端需要与普通 Pulsar 客户端相同的安全配置。将 spring.pulsar.client.authentication 替换为 spring.pulsar.admin.authentication，即可使用上述身份验证配置。

要在启动时创建主题，请添加一个 PulsarTopic 类型的 Bean。如果主题已经存在，该 Bean 将被忽略。

发送消息

Spring 的 PulsarTemplate 是自动配置的，你可以用它来发送消息，如下例所示：

import org.apache.pulsar.client.api.PulsarClientException;

import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.stereotype.Component;

@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");
    }

}

PulsarTemplate 依赖 PulsarProducerFactory 来创建底层 Pulsar 生产者。Spring Boot 自动配置也提供了这个生产者工厂，默认情况下，它会缓存所创建的生产者。你可以通过指定以 spring.pulsar.producer.* 和 spring.pulsar.producer.cache.* 为前缀的应用程序属性来配置生产者工厂和缓存设置。
如果需要对生产者工厂配置进行更多控制，可以考虑注册一个或多个 ProducerBuilderCustomizer Bean。这些定制器将应用于所有创建的生产者。您也可以在发送消息时传递一个 ProducerBuilderCustomizer，以便只影响当前的生产者。
如果需要对发送的消息进行更多控制，可以在发送消息时加入TypedMessageBuilderCustomizer。

反应式发送信息

激活 Reactive 自动配置后，Spring 的 ReactivePulsarTemplate 就会自动配置，你可以用它来发送消息，如下例所示：

import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import org.springframework.stereotype.Component;

@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();
    }

}

ReactivePulsarTemplate 依赖 ReactivePulsarSenderFactory 来实际创建底层发送器。Spring Boot 自动配置也提供了该发送器工厂，默认情况下，该工厂会缓存其创建的生产者。你可以通过指定任何以 spring.pulsar.producer.* 和 spring.pulsar.producer.cache.* 为前缀的应用程序属性来配置发送器工厂和缓存设置。
如果您需要对发送器工厂配置进行更多控制，可以考虑注册一个或多个 ReactiveMessageSenderBuilderCustomizer Bean。这些自定义器将应用于所有创建的发送器。您也可以在发送消息时传入 ReactiveMessageSenderBuilderCustomizer，以便只影响当前的发送者。
如果需要对发送的消息进行更多控制，可以在发送消息时传入 MessageSpecBuilderCustomizer。

接收消息

当 Apache Pulsar 基础架构存在时，任何 bean 都可以用 @PulsarListener 进行注解，以创建一个监听器端点。下面的组件在 someTopic 主题上创建了一个监听器端点：

import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @PulsarListener(topics = "someTopic")
    public void processMessage(String content) {
        // ...
    }

}

Spring Boot 自动配置提供了 PulsarListener 所需的所有组件，如 PulsarListenerContainerFactory 和用于构建底层 Pulsar 消费者的消费者工厂。你可以通过指定任何以 spring.pulsar.listener.* 和 spring.pulsar.consumer.* 为前缀的应用程序属性来配置这些组件。
如果需要对消费者工厂配置进行更多控制，可以考虑注册一个或多个 ConsumerBuilderCustomizer Bean。这些自定义器将应用于工厂创建的所有消费者，因此也会应用于所有 @PulsarListener 实例。您还可以通过设置 @PulsarListener 注解的 consumerCustomizer 属性来自定义单个监听器。

反应式接收消息

当 Apache Pulsar 基础架构存在并激活了反应式自动配置时，任何 bean 都可以用 @ReactivePulsarListener 进行注解，以创建一个反应式监听器端点。下面的组件在 someTopic 主题上创建了一个反应式监听器端点：

import reactor.core.publisher.Mono;

import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @ReactivePulsarListener(topics = "someTopic")
    public Mono<Void> processMessage(String content) {
        // ...
        return Mono.empty();
    }

}

Spring Boot 自动配置提供了 ReactivePulsarListener 所需的所有组件，如 ReactivePulsarListenerContainerFactory 和用于构建底层反应式 Pulsar 消费者的消费者工厂。你可以通过指定任何以 spring.pulsar.listener. 和 spring.pulsar.consumer. 为前缀的应用程序属性来配置这些组件。
如果需要对消费者工厂配置进行更多控制，可以考虑注册一个或多个 ReactiveMessageConsumerBuilderCustomizer Bean。这些自定义器将应用于工厂创建的所有消费者，因此也会应用于所有 @ReactivePulsarListener 实例。您还可以通过设置 @ReactivePulsarListener 注解的 consumerCustomizer 属性来自定义单个监听器。

读取消息

Pulsar 阅读器界面可让应用程序手动管理光标。当您使用阅读器连接到主题时，您需要指定阅读器在连接到主题时从哪条消息开始读取。
当 Apache Pulsar 基础架构存在时，任何 bean 都可以用 @PulsarReader 进行注解，以便使用阅读器读取消息。下面的组件创建了一个阅读器端点，从 someTopic 主题的开头开始阅读消息：

import org.springframework.pulsar.annotation.PulsarReader;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

    @PulsarReader(topics = "someTopic", startMessageId = "earliest")
    public void processMessage(String content) {
        // ...
    }

}

@PulsarReader 依赖 PulsarReaderFactory 来创建底层 Pulsar 阅读器。Spring Boot 自动配置提供了该阅读器工厂，可通过设置任何以 spring.pulsar.reader.* 为前缀的应用程序属性对其进行自定义。
如果需要对阅读器工厂配置进行更多控制，可以考虑注册一个或多个 ReaderBuilderCustomizer Bean。这些自定义器将应用于工厂创建的所有阅读器，因此也会应用于所有 @PulsarReader 实例。您还可以通过设置 @PulsarReader 注解的 readerCustomizer 属性来自定义单个监听器。

反应式读取消息

当 Apache Pulsar 基础架构存在并激活了反应式自动配置时，Spring 就会提供 ReactivePulsarReaderFactory，你可以用它来创建一个阅读器，以便以反应式方式读取消息。下面的组件使用提供的工厂创建了一个阅读器，并从 someTopic 主题中读取了 5 分钟前的一条消息：

import java.time.Instant;
import java.util.List;

import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.reactive.client.api.StartAtSpec;
import reactor.core.publisher.Mono;

import org.springframework.pulsar.reactive.core.ReactiveMessageReaderBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory;
import org.springframework.stereotype.Component;

@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();
        // ...
    }

}

Spring Boot 自动配置提供了这个阅读器工厂，可以通过设置任何以 spring.pulsar.reader.* 为前缀的应用程序属性来对其进行自定义。
如果需要对阅读器工厂的配置进行更多控制，可以考虑在使用该工厂创建阅读器时传递一个或多个 ReactiveMessageReaderBuilderCustomizer 实例。
如果需要对阅读器工厂配置进行更多控制，可以考虑注册一个或多个 ReactiveMessageReaderBuilderCustomizer Bean。这些自定义器将应用于所有创建的阅读器。您也可以在创建阅读器时传递一个或多个 ReactiveMessageReaderBuilderCustomizer，以便只将自定义应用于创建的阅读器。

有关上述任何组件的更多详情以及其他可用功能，请参阅 Spring for Apache Pulsar 参考文档。

其他的 Pulsar 属性

附录的 “集成属性” 部分显示了自动配置所支持的属性。请注意，在大多数情况下，这些属性（hyphenated 或 camelCase）直接映射到 Apache Pulsar 配置属性。详情请参见 Apache Pulsar 文档。
只有 Pulsar 支持的一部分属性可直接通过 PulsarProperties 类使用。如果你想用其他不被直接支持的属性来调整自动配置组件，可以使用上述每个组件所支持的自定义器。

RSocket

RSocket 是用于字节流传输的二进制协议。它通过单个连接上的异步消息传递实现对称交互模型。
Spring Framework 的 spring-messaging 模块为客户端和服务器端的 RSocket 请求者和响应者提供支持。更多详情，包括 RSocket 协议概述，请参见 Spring Framework 参考资料中的 RSocket 部分。

RSocket 策略自动配置

Spring Boot 会自动配置一个 RSocketStrategies Bean，为编码和解码 RSocket 有效载荷提供所有必需的基础结构。默认情况下，自动配置将尝试配置以下内容（按顺序）：

使用 Jackson 的 CBOR 编解码器
使用 Jackson 的 JSON 编解码器

spring-boot-starter-rsocket 启动器提供了这两个依赖项。请参阅 “Jackson 支持” 部分了解更多定制可能性。
开发人员可以通过创建实现 RSocketStrategiesCustomizer 接口的 Bean 来定制 RSocketStrategies 组件。请注意，它们的 @Order 非常重要，因为它决定了编解码器的顺序。

RSocket 服务器自动配置

Spring Boot 提供 RSocket 服务器自动配置功能。spring-boot-starter-rsocket 提供了所需的依赖项。
Spring Boot 允许从 WebFlux 服务器通过 WebSocket 公开 RSocket，或建立独立的 RSocket 服务器。这取决于应用程序的类型及其配置。
对于 WebFlux 应用程序（WebApplicationType.REACTIVE 类型），RSocket 服务器只有在符合以下属性时才会插入 Web 服务器：

spring:
  rsocket:
    server:
      mapping-path: "/rsocket"
      transport: "websocket"

只有 Reactor Netty 才支持将 RSocket 插入网络服务器，因为 RSocket 本身就是使用该库构建的。

或者，将 RSocket TCP 或 websocket 服务器作为独立的嵌入式服务器启动。除了依赖性要求外，唯一需要的配置就是为服务器定义端口：

spring:
  rsocket:
    server:
      port: 9898

Spring Messaging RSocket 支持

Spring Boot 将为 RSocket 自动配置 Spring 消息基础架构。
这意味着 Spring Boot 将创建一个 RSocketMessageHandler Bean，用于处理应用程序的 RSocket 请求。

使用 RSocketRequester 调用 RSocket 服务

一旦服务器和客户端之间建立了 RSocket 通道，任何一方都可以向另一方发送或接收请求。
作为服务器，您可以在 RSocket @Controller 的任何处理程序方法中注入一个 RSocketRequester 实例。作为客户端，您需要先配置并建立 RSocket 连接。在这种情况下，Spring Boot 会使用预期的编解码器自动配置 RSocketRequester.Builder 并应用任何 RSocketConnectorConfigurer Bean。
RSocketRequester.Builder 实例是一个prototype bean，这意味着每个注入点都会提供一个新实例。这样做是有目的的，因为该生成器是有状态的，您不应该使用同一个实例创建具有不同设置的请求器。
下面的代码展示了一个典型示例：

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);
    }

}

Spring Integration

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:
    jdbc:
      initialize-schema: "always"

如果 spring-integration-rsocket 可用，开发人员可使用 "spring.rsocket.server.*" 属性配置 RSocket 服务器，并让它使用 IntegrationRSocketEndpoint 或 RSocketOutboundGateway 组件来处理传入的 RSocket 消息。该基础架构可处理 Spring Integration RSocket 通道适配器和 @MessageMapping 处理程序（已配置 "spring.integration.rsocket.server.message-mapping-enabled"）。
Spring Boot 还可以使用配置属性自动配置 ClientRSocketConnector：

# Connecting to a RSocket server over TCP
spring:
  integration:
    rsocket:
      client:
        host: "example.org"
        port: 9898

# Connecting to a RSocket Server over WebSocket
spring:
  integration:
    rsocket:
      client:
        uri: "ws://example.org"

详情请查看 IntegrationAutoConfiguration 和 IntegrationProperties 类。

WebSockets

Spring Boot 为嵌入式 Tomcat、Jetty 和 Undertow 提供了 WebSockets 自动配置。如果将 war 文件部署到独立容器，Spring Boot 会假定容器负责配置其 WebSocket 支持。
Spring Framework 为 MVC Web 应用程序提供了丰富的 WebSocket 支持，可通过 spring-boot-starter-websocket 模块轻松访问。
反应式网络应用程序也可使用 WebSocket 支持，但需要在 spring-boot-starter-webflux 中包含 WebSocket API：

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

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/195183.html

Spring Boot Messaging中文文档

Spring Boot Messaging中文文档

引言

JMS

ActiveMQ “Classic” 支持

ActiveMQ Artemis支持

使用 JNDI ConnectionFactory

发送消息

接收消息

AMQP

RabbitMQ 支持

发送消息

向流发送消息

接收消息

Apache Kafka 支持

发送消息

接收消息

Kafka Streams

其他的 Kafka 属性

使用嵌入式 Kafka 进行测试

Apache Pulsar 支持

连接 Pulsar

Authentication

SSL

反应式连接 Pulsar

连接 Pulsar 管理器

Authentication

发送消息

反应式发送信息

接收消息

反应式接收消息

读取消息

反应式读取消息

其他的 Pulsar 属性

RSocket

RSocket 策略自动配置

RSocket 服务器自动配置

Spring Messaging RSocket 支持

使用 RSocketRequester 调用 RSocket 服务

Spring Integration

WebSockets

相关推荐

发表回复