跳到主要内容

Apache Pulsar 支持

DeepSeek V3 中英对照 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 也会为响应式组件执行相同的操作。

为了方便地收集命令式和响应式使用的依赖项，分别提供了 spring-boot-starter-pulsar 和 spring-boot-starter-pulsar-reactive 启动器。

连接 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 和插件所需的任何参数来指定使用哪个身份验证插件。您可以将参数设置为参数名称到参数值的映射。以下示例展示了如何配置 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
properties
备注

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

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

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

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。

认证​

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

提示

要在启动时创建一个主题，请添加一个类型为 PulsarTopic 的 bean。如果主题已经存在，该 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");
    }

}
java

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

}
java

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

}
java

Spring Boot 自动配置为 PulsarListener 提供了所有必要的组件，例如 PulsarListenerContainerFactory 以及用于构建底层 Pulsar 消费者的消费者工厂。你可以通过指定任何以 spring.pulsar.listener.* 和 spring.pulsar.consumer.* 为前缀的应用程序属性来配置这些组件。

如果你需要对消费者工厂的配置进行更细致的控制，可以考虑注册一个或多个 ConsumerBuilderCustomizer bean。这些自定义器会应用到工厂创建的所有消费者上，因此也会影响到所有的 @PulsarListener 实例。你也可以通过设置 @PulsarListener 注解的 consumerCustomizer 属性来单独定制某个监听器。

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

以响应式方式接收消息​

当 Apache Pulsar 基础设施存在且 Reactive 自动配置被激活时，任何 Bean 都可以通过 @ReactivePulsarListener 注解来创建一个 Reactive 监听器端点。以下组件在 someTopic 主题上创建了一个 Reactive 监听器端点：

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

}
java

Spring Boot 自动配置为 ReactivePulsarListener 提供了所有必要的组件，例如 ReactivePulsarListenerContainerFactory 以及它用于构建底层 Reactive 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 主题的开头开始读取消息：

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

}
java

@PulsarReader 依赖于 PulsarReaderFactory 来创建底层的 Pulsar reader。Spring Boot 自动配置提供了这个 reader factory，可以通过设置任何以 spring.pulsar.reader.* 为前缀的应用属性来自定义它。

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

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

反应式读取消息​

当 Apache Pulsar 基础设施存在且 Reactive 自动配置被激活时，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();
        // ...
    }

}
java

Spring Boot 自动配置提供了这个读取器工厂，可以通过设置任何以 spring.pulsar.reader.* 为前缀的应用属性来进行自定义。

如果你需要对 reader 工厂配置进行更多控制，请考虑在使用该工厂创建 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。

额外的 Pulsar 属性​

自动配置支持的属性在附录中的 集成属性 部分列出。需要注意的是，这些属性（无论是带连字符的格式还是驼峰格式）大部分都直接映射到 Apache Pulsar 的配置属性。详情请参阅 Apache Pulsar 文档。

只有部分 Pulsar 支持的属性可以通过 PulsarProperties 类直接使用。如果你希望通过额外的属性来调整自动配置的组件，而这些属性并未直接支持，你可以使用上述每个组件所支持的定制器。