|
对于最新稳定版本，请使用 Spring Boot 4.0.4！ |
Apache Pulsar 支持
Apache Pulsar 通过提供对 Spring for Apache Pulsar 项目的自动配置来获得支持。
当类路径中存在 org.springframework.pulsar:spring-pulsar 时，Spring Boot 将自动配置并注册 Apache Pulsar 的经典（命令式）Spring 组件。
当类路径中存在 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。
身份验证
要连接到需要身份验证的 Pulsar 集群，您需要通过设置 pluginClassName 来指定要使用的身份验证插件，以及该插件所需的任何参数。
您可以将这些参数设置为一个映射（map），其中键为参数名称，值为对应的参数值。
以下示例展示了如何配置 AuthenticationOAuth2 插件。
-
Properties
-
YAML
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-instancespring:
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|
您需要确保在 例如，如果你想为 这种缺乏宽松绑定的特性也使得使用环境变量来设置认证参数变得有问题，因为在转换过程中会丢失大小写敏感性。 如果你使用环境变量来设置这些参数，那么你需要按照 Apache Pulsar 的 Spring 参考文档中的这些步骤进行操作，才能使其正常工作。 |
以响应式方式连接 Pulsar
当响应式自动配置被激活时，Spring Boot 将自动配置并注册一个 ReactivePulsarClient Bean。
ReactivePulsarClient 会适配前面描述的 PulsarClient 的一个实例。
因此，请遵循上一节的说明，配置由 ReactivePulsarClient 使用的 PulsarClient。
连接到 Pulsar 管理端
Spring for Apache Pulsar 的 PulsarAdministration 客户端也已自动配置。
默认情况下，应用程序会尝试连接到位于 http://localhost:8080 的本地 Pulsar 实例。
可以通过将 spring.pulsar.admin.service-url 属性设置为格式为 (http|https)://<host>:<port> 的其他值来进行调整。
如果您需要对配置进行更多控制，请考虑注册一个或多个 PulsarAdminBuilderCustomizer Bean。
身份验证
当访问需要身份验证的 Pulsar 集群时，管理客户端需要与常规 Pulsar 客户端相同的安全部分配置。
您可以通过将 #messaging.pulsar.connecting.auth 替换为 spring.pulsar.client.authentication 来使用前述的身份验证配置。
要在启动时创建主题，请添加一个类型为 PulsarTopic 的 Bean。
如果该主题已存在，则该 Bean 将被忽略。 |
发送消息
Spring 的 PulsarTemplate 是自动配置的，您可以使用它来发送消息，如下例所示：
-
Java
-
Kotlin
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() {
this.pulsarTemplate.send("someTopic", "Hello");
}
}import org.apache.pulsar.client.api.PulsarClientException
import org.springframework.pulsar.core.PulsarTemplate
import org.springframework.stereotype.Component
@Component
class MyBean(private val pulsarTemplate: PulsarTemplate<String>) {
@Throws(PulsarClientException::class)
fun someMethod() {
pulsarTemplate.send("someTopic", "Hello")
}
}PulsarTemplate 依赖于 PulsarProducerFactory 来创建底层的 Pulsar 生产者。
Spring Boot 自动配置也提供了该生产者工厂，默认情况下，它会缓存所创建的生产者。
您可以通过指定任何以 spring.pulsar.producer.* 和 spring.pulsar.producer.cache.* 为前缀的应用程序属性来配置生产者工厂和缓存设置。
如果您需要对生产者工厂配置进行更多控制，可以考虑注册一个或多个 ProducerBuilderCustomizer Bean。
这些定制器将应用于所有创建的生产者。
您还可以在发送消息时传入一个 ProducerBuilderCustomizer，以仅影响当前生产者。
如果您需要对发送的消息进行更多控制，可以在发送消息时传入一个 TypedMessageBuilderCustomizer。
以响应式方式发送消息
当 Reactive 自动配置被激活时，Spring 的 ReactivePulsarTemplate 会被自动配置，您可以使用它来发送消息，如下例所示：
-
Java
-
Kotlin
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();
}
}import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate
import org.springframework.stereotype.Component
@Component
class MyBean(private val pulsarTemplate: ReactivePulsarTemplate<String>) {
fun someMethod() {
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 主题上创建了一个监听器端点：
-
Java
-
Kotlin
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.stereotype.Component;
@Component
public class MyBean {
@PulsarListener(topics = "someTopic")
public void processMessage(String content) {
// ...
}
}import org.springframework.pulsar.annotation.PulsarListener
import org.springframework.stereotype.Component
@Component
class MyBean {
@PulsarListener(topics = ["someTopic"])
fun processMessage(content: String?) {
// ...
}
}Spring Boot 自动配置提供了 PulsarListener 所需的所有组件，例如 PulsarListenerContainerFactory 及其用于构建底层 Pulsar 消费者的消费者工厂。
您可以通过指定任何以 spring.pulsar.listener.* 和 spring.pulsar.consumer.* 为前缀的应用程序属性来配置这些组件。
如果您需要对消费者工厂的配置进行更多控制，可以考虑注册一个或多个 ConsumerBuilderCustomizer Bean。
这些定制器将应用于由该工厂创建的所有消费者，因此也适用于所有 @PulsarListener 实例。
您还可以通过设置 @PulsarListener 注解的 consumerCustomizer 属性来自定义单个监听器。
如果你需要对实际的容器工厂配置进行更精细的控制，请考虑注册一个或多个 PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> Bean。
以响应式方式接收消息
当 Apache Pulsar 基础设施存在且响应式自动配置已激活时，任何 Bean 都可以使用 @ReactivePulsarListener 注解来创建响应式监听器端点。
以下组件在 someTopic 主题上创建了一个响应式监听器端点：
-
Java
-
Kotlin
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();
}
}import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener
import org.springframework.stereotype.Component
import reactor.core.publisher.Mono
@Component
class MyBean {
@ReactivePulsarListener(topics = ["someTopic"])
fun processMessage(content: String?): Mono<Void> {
// ...
return Mono.empty()
}
}Spring Boot 自动配置提供了 ReactivePulsarListener 所需的所有组件，例如 ReactivePulsarListenerContainerFactory 及其用于构建底层响应式 Pulsar 消费者的消费者工厂。
您可以通过指定任何以 spring.pulsar.listener.* 和 spring.pulsar.consumer.* 为前缀的应用程序属性来配置这些组件。
如果您需要对消费者工厂的配置进行更多控制，可以考虑注册一个或多个 ReactiveMessageConsumerBuilderCustomizer Bean。
这些定制器将应用于由该工厂创建的所有消费者，因此也适用于所有 @ReactivePulsarListener 实例。
您还可以通过设置 @ReactivePulsarListener 注解的 consumerCustomizer 属性来自定义单个监听器。
如果你需要对实际的容器工厂配置进行更精细的控制，请考虑注册一个或多个 PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> Bean。
读取消息
Pulsar Reader 接口允许应用程序手动管理游标。 当您使用 Reader 连接到一个主题时，需要指定 Reader 在连接到该主题时从哪条消息开始读取。
当 Apache Pulsar 基础设施存在时，任何 Bean 都可以使用 @PulsarReader 进行注解，以通过 reader 消费消息。
以下组件创建一个 reader 端点，该端点从 someTopic 主题的开头开始读取消息：
-
Java
-
Kotlin
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) {
// ...
}
}import org.springframework.pulsar.annotation.PulsarReader
import org.springframework.stereotype.Component
@Component
class MyBean {
@PulsarReader(topics = ["someTopic"], startMessageId = "earliest")
fun processMessage(content: String?) {
// ...
}
}@PulsarReader 依赖 PulsarReaderFactory 来创建底层的 Pulsar 读取器。
Spring Boot 自动配置提供了该读取器工厂，可以通过设置任何以 spring.pulsar.reader.* 为前缀的应用程序属性来自定义它。
如果您需要对读取器工厂的配置进行更精细的控制，可以考虑注册一个或多个 ReaderBuilderCustomizer Bean。
这些定制器将应用于由该工厂创建的所有读取器，因此也适用于所有 @PulsarReader 实例。
您还可以通过设置 @PulsarReader 注解的 readerCustomizer 属性来定制单个监听器。
如果你需要对实际的容器工厂配置进行更精细的控制，请考虑注册一个或多个 PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> Bean。
以响应式方式读取消息
当 Apache Pulsar 基础设施存在且响应式自动配置已激活时，将提供 Spring 的 ReactivePulsarReaderFactory，您可以使用它创建一个读取器，以响应式方式读取消息。
以下组件使用提供的工厂创建一个读取器，并从 someTopic 主题中读取 5 分钟前的一条消息：
-
Java
-
Kotlin
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();
// ...
}
}import org.apache.pulsar.client.api.Schema
import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder
import org.apache.pulsar.reactive.client.api.StartAtSpec
import org.springframework.pulsar.reactive.core.ReactiveMessageReaderBuilderCustomizer
import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory
import org.springframework.stereotype.Component
import java.time.Instant
@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 自动配置提供了此读取器工厂，可通过设置任意以 spring.pulsar.reader.* 为前缀的应用程序属性对其进行自定义。
如果您需要对读取器工厂配置进行更多控制，请考虑在使用该工厂创建读取器时传入一个或多个 ReactiveMessageReaderBuilderCustomizer 实例。
如果您需要对读取器工厂配置进行更多控制，可以考虑注册一个或多个 ReactiveMessageReaderBuilderCustomizer Bean。
这些自定义器将应用于所有创建的读取器。
您也可以在创建读取器时传递一个或多个 ReactiveMessageReaderBuilderCustomizer，以便仅将自定义设置应用于所创建的读取器。
| 有关上述任一组件的更多详细信息，以及了解其他可用功能，请参阅 Apache Pulsar 的 Spring 参考文档。 |
事务支持
Spring for Apache Pulsar 在使用 PulsarTemplate 和 @PulsarListener 时支持事务。
| 使用响应式变体时，当前不支持事务。 |
将 spring.pulsar.transaction.enabled 属性设置为 true 将会：
-
配置一个
PulsarTransactionManagerBean -
为
PulsarTemplate启用事务支持 -
为
@PulsarListener方法启用事务支持
@PulsarListener 的 transactional 属性可用于微调事务与监听器的使用时机。
为了更灵活地控制 Spring for Apache Pulsar 的事务功能，您应当定义自己的 PulsarTemplate 和/或 ConcurrentPulsarListenerContainerFactory Bean。
如果默认自动配置的 PulsarTransactionManager 不适用，您也可以定义一个 PulsarAwareTransactionManager Bean。
额外的 Pulsar 属性
自动配置所支持的属性列在附录的集成属性部分。 请注意，这些属性（使用连字符或驼峰命名法）在大多数情况下直接映射到 Apache Pulsar 的配置属性。 有关详细信息，请参阅 Apache Pulsar 官方文档。
只有 Pulsar 支持的部分属性可以通过 PulsarProperties 类直接使用。
如果您希望使用其他未直接支持的属性来调整自动配置的组件，可以使用上述每个组件所支持的自定义器。