|
对于最新的稳定版本，请使用 Spring Boot 3.5.5！ |
Apache Pulsar 支持
Apache Pulsar 通过提供 Spring for Apache Pulsar 项目的自动配置来支持。
Spring Boot 将自动配置和注册经典（命令式）Spring for Apache Pulsar 组件，当org.springframework.pulsar:spring-pulsar在类路径上。
当org.springframework.pulsar:spring-pulsar-reactive在类路径上。
有spring-boot-starter-pulsar和spring-boot-starter-pulsar-reactivestarter 分别方便地收集命令式和响应式使用的依赖项。
连接到 Pulsar
当您使用 Pulsar Starters时，Spring Boot 将自动配置并注册一个PulsarClient豆。
默认情况下，应用程序会尝试连接到本地 Pulsar 实例pulsar://localhost:6650.
这可以通过设置spring.pulsar.client.service-url属性设置为不同的值。
| 该值必须是有效的 Pulsar 协议 URL |
您可以通过指定任何spring.pulsar.client.*以前缀为前缀的应用程序属性。
如果您需要对配置进行更多控制，请考虑注册一个或多个PulsarClientBuilderCustomizer豆。
认证
要连接到需要身份验证的 Pulsar 集群，您需要通过设置pluginClassName以及插件所需的任何参数。
您可以将参数设置为参数名称与参数值的映射。
以下示例显示如何配置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|
您需要确保在 例如，如果要为 这种缺乏宽松的绑定也使得将环境变量用于身份验证参数成为问题，因为在翻译过程中会丢失区分大小写。 如果您将环境变量用于参数，则需要按照 Spring for Apache Pulsar 参考文档中的这些步骤进行作，才能使其正常工作。 |
响应式连接到 Pulsar
激活响应式自动配置后，Spring Boot 将自动配置并注册一个ReactivePulsarClient豆。
这ReactivePulsarClient调整了前面描述的PulsarClient.
因此，请按照上一节配置PulsarClient由ReactivePulsarClient.
连接到 Pulsar 管理
Apache Pulsar 的 SpringPulsarAdministration客户端也是自动配置的。
默认情况下，应用程序会尝试连接到本地 Pulsar 实例http://localhost:8080.
这可以通过设置spring.pulsar.admin.service-url属性设置为表单中的不同值(http|https)://<host>:<port>.
如果您需要对配置进行更多控制，请考虑注册一个或多个PulsarAdminBuilderCustomizer豆。
认证
当访问需要身份验证的 Pulsar 集群时，管理客户端需要与常规 Pulsar 客户端相同的安全配置。您可以通过将spring.pulsar.client.authentication跟spring.pulsar.admin.authentication.
要在启动时创建主题，请添加类型为PulsarTopic. 如果主题已经存在，则忽略 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豆。 这些定制器将应用于所有创建的生产者。您还可以传入ProducerBuilderCustomizer发送消息时，仅影响当前生产者。
如果您需要对正在发送的消息进行更多控制，您可以将TypedMessageBuilderCustomizer发送消息时。
被动发送消息
激活响应式自动配置后，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豆。 这些定制器将应用于所有创建的发件人。您还可以传入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豆。 这些定制器适用于工厂创建的所有消费者，因此适用于所有@PulsarListener实例。 您还可以通过设置consumerCustomizer属性的@PulsarListener注解。
如果您需要对实际的容器工厂配置进行更多控制，请考虑注册一个或多个PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>>豆。
被动接收消息
当 Apache Pulsar 基础设施存在并且激活了 Reactive 自动配置时，任何 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豆。 这些定制器适用于工厂创建的所有消费者，因此适用于所有@ReactivePulsarListener实例。 您还可以通过设置consumerCustomizer属性的@ReactivePulsarListener注解。
如果您需要对实际的容器工厂配置进行更多控制，请考虑注册一个或多个PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>>豆。
阅读消息
Pulsar 阅读器界面使应用程序能够手动管理游标。 使用阅读器连接到主题时，需要指定读取器在连接到主题时从哪条消息开始阅读。
当 Apache Pulsar 基础设施存在时，任何 bean 都可以使用@PulsarReader使用阅读器使用消息。
以下组件创建一个读取器端点，该端点从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豆。
这些定制器适用于工厂创建的所有阅读器，因此适用于所有@PulsarReader实例。 您还可以通过设置readerCustomizer属性的@PulsarReader注解。
如果您需要对实际的容器工厂配置进行更多控制，请考虑注册一个或多个PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>>豆。
被动地阅读消息
当 Apache Pulsar 基础设施存在并且激活了 Reactive 自动配置时，Spring 的ReactivePulsarReaderFactory，您可以使用它来创建阅读器，以便以反应方式阅读消息。
以下组件使用提供的工厂创建一个读取器，并从someTopic主题：
-
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豆。
这些定制器将应用于所有创建的读者。
您还可以通过一个或多个ReactiveMessageReaderBuilderCustomizer创建阅读器时，仅将自定义项应用于创建的阅读器。
| 有关上述任何组件的更多详细信息以及发现其他可用功能，请参阅 Spring for Apache Pulsar 参考文档。 |
交易支持
Spring for Apache Pulsar 在使用PulsarTemplate和@PulsarListener.
| 使用响应式变体时，当前不支持事务。 |
设置spring.pulsar.transaction.enabled属性设置为true将：
-
启用事务支持
PulsarTemplate -
启用事务支持
@PulsarListener方法
这transactional属性@PulsarListener可用于微调事务何时应与侦听器一起使用。
为了更好地控制 Spring for Apache Pulsar 事务功能，您应该定义自己的PulsarTemplate和/或ConcurrentPulsarListenerContainerFactory豆。
您还可以定义一个PulsarAwareTransactionManagerbean 如果默认自动配置PulsarTransactionManager不合适。
其他 Pulsar 属性
自动配置支持的属性显示在附录的集成属性部分中。 请注意，在大多数情况下，这些属性（连字符或驼峰命名法）直接映射到 Apache Pulsar 配置属性。 有关详细信息，请参阅 Apache Pulsar 文档。
只有 Pulsar 支持的属性子集可以直接通过PulsarProperties类。
如果您希望使用不直接支持的其他属性来调整自动配置的组件，您可以使用上述每个组件支持的定制器。