跳到主要内容
版本：3.5.10

Apache Pulsar 支持

QWen Max 中英对照 Apache Pulsar Support

Apache Pulsar 通过提供 Spring for Apache Pulsar 项目的自动配置来支持。

当 org.springframework.pulsar:spring-pulsar 在类路径中时，Spring Boot 会自动配置并注册经典的（命令式）Spring for Apache Pulsar 组件。当 org.springframework.pulsar:spring-pulsar-reactive 在类路径中时，它也会对响应式组件执行相同的操作。

有 spring-boot-starter-pulsar 和 spring-boot-starter-pulsar-reactive 两个 starter，分别用于方便地收集命令式和响应式使用所需的依赖。

连接到 Pulsar​

当你使用 Pulsar starter 时，Spring Boot 将自动配置并注册一个 PulsarClient bean。

默认情况下，应用程序会尝试连接到位于 pulsar://localhost:6650 的本地 Pulsar 实例。可以通过将 spring.pulsar.client.service-url 属性设置为其他值来调整该行为。

备注

该值必须是一个有效的 Pulsar Protocol URL

你可以通过指定任意以 spring.pulsar.client.* 为前缀的应用程序属性来配置客户端。

如果你需要对配置进行更精细的控制，可以考虑注册一个或多个 PulsarClientBuilderCustomizer Bean。

认证​

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

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-instance
备注

你需要确保在 spring.pulsar.client.authentication.param.* 下定义的名称与你的认证插件所期望的名称完全一致（通常为驼峰命名法）。Spring Boot 不会对这些条目尝试任何形式的宽松绑定（relaxed binding）。

例如，如果你想为 AuthenticationOAuth2 认证插件配置 issuer URL，则必须使用 spring.pulsar.client.authentication.param.issuerUrl。如果你使用其他形式，比如 issuerurl 或 issuer-url，该设置将不会被应用到插件中。

这种缺乏宽松绑定的特性也使得使用环境变量来配置认证参数变得有问题，因为在转换过程中会丢失大小写敏感性。如果你使用环境变量来设置这些参数，则需要按照 Spring for Apache Pulsar 参考文档中的 这些步骤 进行操作，以确保其正常工作。

SSL​

默认情况下，Pulsar 客户端以明文方式与 Pulsar 服务通信。你可以按照 Apache Pulsar 的 Spring 参考文档中的这些步骤来启用 TLS 加密。

有关客户端和认证的完整详细信息，请参阅 Apache Pulsar 的 Spring 参考文档。

以响应式方式连接 Pulsar​

当 Reactive 自动配置被激活时，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。

Authentication​

在访问需要身份验证的 Pulsar 集群时，admin 客户端需要与常规 Pulsar 客户端相同的安全部分配置。你可以使用前述的 authentication configuration，只需将 spring.pulsar.client.authentication 替换为 spring.pulsar.admin.authentication 即可。

提示

要在启动时创建一个 topic，请添加一个类型为 PulsarTopic 的 bean。如果该 topic 已存在，则该 bean 会被忽略。

发送消息​

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

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

}

PulsarTemplate 依赖于 PulsarProducerFactory 来创建底层的 Pulsar producer。Spring Boot 的自动配置也会提供该 producer factory，默认情况下，它会缓存所创建的 producers。你可以通过指定任意以 spring.pulsar.producer.* 和 spring.pulsar.producer.cache.* 为前缀的应用程序属性来配置 producer factory 及其缓存设置。

如果你需要对 Producer 工厂的配置进行更精细的控制，可以考虑注册一个或多个 ProducerBuilderCustomizer Bean。这些定制器会应用到所有创建的 Producer 上。你也可以在发送消息时传入一个 ProducerBuilderCustomizer，以仅影响当前的 Producer。

如果你需要对发送的消息进行更精细的控制，可以在发送消息时传入一个 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 来实际创建底层的 sender。Spring Boot 的自动配置也会提供该 sender factory，默认情况下，它会缓存所创建的 producer。你可以通过指定任意以 spring.pulsar.producer.* 和 spring.pulsar.producer.cache.* 为前缀的应用程序属性来配置 sender factory 及其缓存设置。

如果你需要对 sender factory 的配置进行更精细的控制，可以考虑注册一个或多个 ReactiveMessageSenderBuilderCustomizer Bean。这些定制器（customizer）会被应用到所有创建的 sender 上。你也可以在发送消息时传入一个 ReactiveMessageSenderBuilderCustomizer，以仅影响当前的 sender。

如果你需要对发送的消息进行更精细的控制，可以在发送消息时传入一个 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.* 为前缀的应用程序属性来配置这些组件。

如果你需要对消费者工厂（consumer factory）的配置进行更精细的控制，可以考虑注册一个或多个 ConsumerBuilderCustomizer Bean。这些定制器（customizers）会应用于该工厂创建的所有消费者，因此也适用于所有 @PulsarListener 实例。你也可以通过设置 @PulsarListener 注解的 consumerCustomizer 属性，来定制单个监听器。

如果你需要对实际的容器工厂配置进行更多控制，请考虑注册一个或多个 PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> Bean。

以响应式方式接收消息​

当 Apache Pulsar 基础设施存在且 Reactive 自动配置已激活时，任何 Bean 都可以使用 @ReactivePulsarListener 注解来创建一个 reactive listener endpoint。以下组件在 someTopic 主题上创建了一个 reactive listener endpoint：

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 以及它用于构建底层 reactive Pulsar 消费者的消费者工厂。你可以通过指定任意以 spring.pulsar.listener.* 和 spring.pulsar.consumer.* 为前缀的应用程序属性来配置这些组件。

如果你需要对消费者工厂的配置进行更精细的控制，可以考虑注册一个或多个 ReactiveMessageConsumerBuilderCustomizer Bean。这些定制器会应用到该工厂创建的所有消费者上，因此也适用于所有的 @ReactivePulsarListener 实例。你也可以通过设置 @ReactivePulsarListener 注解的 consumerCustomizer 属性来定制单个监听器。

如果你需要对实际的容器工厂配置进行更精细的控制，可以考虑注册一个或多个 PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> Bean。

读取消息​

Pulsar Reader 接口允许应用程序手动管理游标。当使用 Reader 连接到一个 topic 时，需要指定 Reader 在连接到该 topic 时从哪条消息开始读取。

当 Apache Pulsar 基础设施存在时，任何 Bean 都可以使用 @PulsarReader 注解，通过 Reader 来消费消息。以下组件创建了一个 Reader 端点，该端点从 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](https://docs.spring.io/spring-pulsar/docs/1.2.x/api/org/springframework/pulsar/annotation/PulsarReader.html) 依赖于 PulsarReaderFactory 来创建底层的 Pulsar reader。Spring Boot 自动配置提供了该 reader factory，可以通过设置任意以 spring.pulsar.reader.* 为前缀的应用程序属性来自定义。

如果你需要对 reader factory 的配置进行更精细的控制，可以考虑注册一个或多个 ReaderBuilderCustomizer Bean。这些定制器会应用到该工厂创建的所有 reader，因此也会影响到所有的 @PulsarReader 实例。你也可以通过设置 @PulsarReader 注解的 readerCustomizer 属性来定制单个 listener。

如果你需要对实际的容器工厂配置进行更精细的控制，可以考虑注册一个或多个 PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> Bean。

以响应式方式读取消息​

当 Apache Pulsar 基础设施存在且 Reactive 自动配置被激活时，Spring 会提供 ReactivePulsarReaderFactory，你可以使用它以响应式方式创建 reader 来读取消息。以下组件使用所提供的 factory 创建一个 reader，并从 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 自动配置提供了此 reader factory，可以通过设置任意以 spring.pulsar.reader.* 为前缀的应用程序属性对其进行自定义。

如果你需要对 reader factory 的配置进行更精细的控制，可以在使用该 factory 创建 reader 时传入一个或多个 ReactiveMessageReaderBuilderCustomizer 实例。

如果你需要对 reader factory 的配置进行更精细的控制，可以考虑注册一个或多个 ReactiveMessageReaderBuilderCustomizer Bean。这些定制器会应用到所有创建的 reader 上。你也可以在创建 reader 时传入一个或多个 ReactiveMessageReaderBuilderCustomizer，以仅将定制应用于所创建的 reader。

提示

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

事务支持​

Spring for Apache Pulsar 在使用 PulsarTemplate 和 @PulsarListener 时支持事务。

备注

使用响应式变体时，目前不支持事务。

将 spring.pulsar.transaction.enabled 属性设置为 true 将会：

@PulsarListener 的 transactional 属性可用于微调监听器何时应使用事务。

若要更精细地控制 Spring for Apache Pulsar 的事务特性，您应自行定义 PulsarTemplate 和/或 ConcurrentPulsarListenerContainerFactory Bean。如果默认自动配置的 PulsarTransactionManager 不符合您的需求，您还可以定义一个 PulsarAwareTransactionManager Bean。

Additional Pulsar Properties​

自动配置所支持的属性在附录的 Integration Properties 一节中列出。请注意，这些属性（使用连字符或驼峰命名法）在大多数情况下直接映射到 Apache Pulsar 的配置属性。详情请参见 Apache Pulsar 文档。

只有 Pulsar 支持的属性的一个子集可以通过 PulsarProperties 类直接获取。如果你希望通过未被直接支持的额外属性来调整自动配置的组件，可以使用上述每个组件所支持的自定义器（customizer）。