适用于 Apache Pulsar 的 Spring Cloud Stream Binder (Spring Cloud Stream Binder for Apache Pulsar) | Spring Cloud Stream4.1.2中文文档|Spring官方文档|SpringBoot 教程|Spring中文网

Spring for Apache Pulsar 为 Spring Cloud Stream 提供了一个 binder，我们可以使用它来构建使用 pub-sub 范式的事件驱动的微服务。在本节中，我们将介绍该活页夹的基本细节。

用法

我们需要在应用程序上包含以下依赖项，以便将 Apache Pulsar binder 用于 Spring Cloud Stream。

Maven
Gradle

<dependencies>
    <dependency>
        <groupId>org.springframework.pulsar</groupId>
        <artifactId>spring-pulsar-spring-cloud-stream-binder</artifactId>
    </dependency>
</dependencies>

dependencies {
    implementation 'org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder'
}

概述

适用于 Apache Pulsar 的 Spring Cloud Stream 绑定器允许应用程序专注于业务逻辑，而不是处理管理和维护 Pulsar 的较低级别细节。活页夹为应用程序开发人员处理所有这些细节。 Spring Cloud Stream 带来了基于 Spring Cloud Function 的强大编程模型，允许应用程序开发人员使用函数样式编写复杂的事件驱动应用程序。应用程序可以从中间件中立的方式开始，然后通过 Spring Boot 配置属性将 Pulsar 主题映射为 Spring Cloud Stream 中的目标。 Spring Cloud Stream 建立在 Spring Boot 之上，当使用 Spring Cloud Stream 编写事件驱动的微服务时，您实际上是在编写 Boot 应用程序。这是一个简单的 Spring Cloud Stream 应用程序。

@SpringBootApplication
public class SpringPulsarBinderSampleApp {

	private final Logger logger = LoggerFactory.getLogger(this.getClass());

	public static void main(String[] args) {
		SpringApplication.run(SpringPulsarBinderSampleApp.class, args);
	}

	@Bean
	public Supplier<Time> timeSupplier() {
		return () -> new Time(String.valueOf(System.currentTimeMillis()));
	}

	@Bean
	public Function<Time, EnhancedTime> timeProcessor() {
		return (time) -> {
			EnhancedTime enhancedTime = new EnhancedTime(time, "5150");
			this.logger.info("PROCESSOR: {} --> {}", time, enhancedTime);
			return enhancedTime;
		};
	}

	@Bean
	public Consumer<EnhancedTime> timeLogger() {
		return (time) -> this.logger.info("SINK:      {}", time);
	}

	record Time(String time) {
	}

	record EnhancedTime(Time time, String extra) {
	}

}

上面的示例应用程序，一个完整的Spring Boot应用程序，值得一些解释。但是，在第一次通过时，您可以看到这只是普通的 Java 和一些 Spring 和 Spring Boot 注解。我们这里有三种方法 - a ， a ，最后是 a 。供应商以毫秒为单位生成当前时间，函数占用此时间，然后通过添加一些随机数据来增强它，然后消费者记录增强的时间。Beanjava.util.function.Supplierjava.util.function.Functionjava.util.function.Consumer

为了简洁起见，我们省略了所有导入，但在整个应用程序中没有特定于 Spring Cloud Stream 的内容。它如何成为与 Apache Pulsar 交互的 Spring Cloud Stream 应用程序？您必须在应用程序中包含活页夹的上述依赖项。添加该依赖项后，必须提供以下配置属性。

spring:
  cloud:
    function:
      definition: timeSupplier;timeProcessor;timeLogger;
    stream:
      bindings:
        timeProcessor-in-0:
          destination: timeSupplier-out-0
        timeProcessor-out-0:
          destination: timeProcessor-out-0
        timeLogger-in-0:
          destination: timeProcessor-out-0

至此，上面的Spring Boot应用就成为了基于Spring Cloud Stream的端到端事件驱动应用。因为我们在类路径上有 Pulsar 绑定器，所以应用程序与 Apache Pulsar 交互。如果应用程序中只有一个函数，那么我们不需要告诉 Spring Cloud Stream 激活该函数进行执行，因为它默认执行。如果应用程序中有多个这样的函数，就像我们的示例一样，我们需要指示 Spring Cloud Stream 我们想要激活哪些函数。在我们的例子中，我们需要激活所有这些，我们通过属性来做到这一点。默认情况下，Bean 名称将成为 Spring Cloud Stream 绑定名称的一部分。绑定是Spring Cloud Stream中一个基本抽象的概念，框架使用它与中间件目标进行通信。 Spring Cloud Stream 所做的几乎所有事情都发生在具体的绑定上。供应商只有输出绑定;函数具有输入和输出绑定，而使用者只有输入绑定。让我们以供应商 bean 为例 - 此供应商的默认绑定名称为。同样，函数的默认绑定名称将位于入站和出站上。有关如何更改默认绑定名称的详细信息，请参阅 Spring Cloud Stream 参考文档。在大多数情况下，使用默认绑定名称就足够了。我们在绑定名称上设置目标，如上所示。如果未提供目标，则绑定名称将成为目标的值，如的情况所示。spring.cloud.function.definitiontimeSupplier.timeSupplier-out-0timeProcessortimeProcessor-in-0timeProcessor-out-0timeSupplier-out-0

运行上述应用程序时，您应该看到供应商每秒执行一次，然后由函数消耗并增加记录器使用者消耗的时间。

基于活页夹的应用程序中的消息转换

在上面的示例应用程序中，我们没有提供用于消息转换的架构信息。这是因为，默认情况下，Spring Cloud Stream 通过 Spring Messaging 项目使用 Spring Framework 中建立的消息传递支持来使用其消息转换机制。除非指定，否则 Spring Cloud Stream 将用作入站和出站绑定上的消息转换。在出站时，数据被序列化为然后 Pulsar 绑定器通过网络将其发送到 Pulsar 主题。同样，在入站时，数据将作为 Pulsar 主题使用，然后使用适当的消息转换器转换为目标类型。application/jsoncontent-typebyte[],Schema.BYTESbyte[]

使用 Pulsar Schema 在 Pulsar 中使用原生转换

尽管默认使用框架提供的消息转换，但 Spring Cloud Stream 允许每个绑定器确定消息的转换方式。假设应用程序选择走这条路线。在这种情况下，Spring Cloud Stream 避免使用任何 Spring 提供的消息转换工具，并传递它接收或生成的数据。 Spring Cloud Stream 中的此功能在生产者端称为原生编码，在消费者端称为原生解码。这意味着编码和解码原生发生在目标中间件上，在我们的例子中，发生在 Apache Pulsar 上。对于上面的应用，我们可以使用以下配置来绕过框架转换，并使用原生编码和解码。

spring:
  cloud:
    stream:
      bindings:
        timeSupplier-out-0:
          producer:
            use-native-encoding: true
        timeProcessor-in-0:
          destination: timeSupplier-out-0
          consumer:
            use-native-decoding: true
        timeProcessor-out-0:
          destination: timeProcessor-out-0
          producer:
            use-native-encoding: true
        timeLogger-in-0:
          destination: timeProcessor-out-0
          consumer:
            use-native-decoding: true
      pulsar:
        bindings:
          timeSupplier-out-0:
            producer:
              schema-type: JSON
              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
          timeProcessor-in-0:
            consumer:
              schema-type: JSON
              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
          timeProcessor-out-0:
            producer:
              schema-type: AVRO
              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
          timeLogger-in-0:
            consumer:
              schema-type: AVRO
              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime

用于在生产者端启用本机编码的属性是核心 Spring Cloud Stream 的绑定级别属性。在生产者绑定上设置它 - 并将其设置为类似地，使用 - 用于消费者绑定，并将其设置为如果我们决定使用原生编码和解码，在 Pulsar 的情况下，我们需要设置相应的模式和底层消息类型信息。此信息作为扩展绑定属性提供。如上图所示，属性为 - 用于架构信息和实际目标类型。如果消息上同时具有键和值，则可以使用和来指定其目标类型。spring.cloud.stream.bindings.<binding-name>.producer.use-native-encodingtrue.spring.cloud.stream.bindings.<binding-name>.consumer.user-native-decodingtrue.spring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.schema-typespring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.message-typemessage-key-typemessage-value-type

省略该属性时，将查阅任何已配置的自定义架构映射。schema-type

邮件头转换

每条消息通常都有标头信息，当消息通过Spring Cloud Stream输入和输出绑定在Pulsar和Spring Messaging之间遍历时，需要携带这些信息。为了支持这种遍历，框架处理必要的消息标头转换。

Pulsar 接头

自定义标头映射器

Pulsar 绑定器配置了一个默认的标头映射器，可以通过提供你自己的 bean 来覆盖它。PulsarHeaderMapper

在以下示例中，JSON 标头映射器配置为：

映射所有入站标头（键为“top”或“secret”的标头除外）
映射出站标头（带有键“id”、“timestamp”或“userId”的标头除外）
仅信任“com.acme”包中的对象进行出站反序列化
使用简单编码对任何“com.acme.Money”标头值进行反/序列化toString()

@Bean
public PulsarHeaderMapper customPulsarHeaderMapper() {
    return JsonPulsarHeaderMapper.builder()
            .inboundPatterns("!top", "!secret", "*")
            .outboundPatterns("!id", "!timestamp", "!userId", "*")
            .trustedPackages("com.acme")
            .toStringClasses("com.acme.Money")
            .build();
}

省略该属性时，将查阅任何已配置的自定义架构映射。schema-type

在 Binder 中使用 Pulsar 属性

该绑定器使用 Spring for Apache Pulsar 框架中的基本组件来构建其生产者和使用者绑定。由于基于 binder 的应用程序是 Spring Boot 应用程序，因此 binder 默认使用 Spring for Apache Pulsar 的 Spring Boot 自动配置。因此，所有在核心框架级别可用的 Pulsar Spring Boot 属性也可以通过 binder 获得。例如，可以使用带有前缀等的属性。此外，您还可以在绑定器级别设置这些 Pulsar 属性。例如，这也将起作用 - 或 .spring.pulsar.producer…spring.pulsar.consumer…spring.cloud.stream.pulsar.binder.producer…spring.cloud.stream.pulsar.binder.consumer…

上述任何一种方法都可以，但是当使用这样的属性时，它会应用于整个应用程序。如果应用程序中有多个函数，则它们都具有相同的属性。您还可以在扩展绑定属性级别设置这些 Pulsar 属性来解决此问题。扩展绑定属性应用于绑定本身。例如，如果你有一个输入和输出绑定，并且都需要一组单独的 Pulsar 属性，那么你必须在扩展绑定上设置它们。生产者绑定的模式是。同样，对于使用者绑定，模式为。这样，您就可以将一组单独的 Pulsar 属性应用于同一应用程序中的不同绑定。spring.cloud.stream.pulsar.bindings.<output-binding-name>.producer…spring.cloud.stream.pulsar.bindings.<input-binding-name>.consumer…

最高优先级是扩展绑定属性。在绑定器中应用属性的优先顺序是（从最高到最低）。extended binding properties → binder properties → Spring Boot properties.

Pulsar Binder 属性资源

以下是一些资源，可以用来查找有关通过 Pulsar 活页夹提供的属性的更多信息。

Pulsar 生产者绑定配置。这些属性需要前缀。 Spring Boot 提供的所有 Pulsar 生产者属性也可以通过此配置类获得。spring.cloud.stream.bindings.<binding-name>.producer

Pulsar 消费者绑定配置。这些属性需要前缀。 Spring Boot 提供的所有 Pulsar 使用者属性也可以通过此配置类获得。spring.cloud.stream.bindings.<binding-name>.consumer

有关特定于 Pulsar 绑定器的常见配置属性，请参阅此处。这些属性需要前缀。上面指定的生产者和使用者属性（包括 Spring Boot 属性）可以在活页夹中使用 or 前缀。spring.cloud.stream.pulsar.binderspring.cloud.stream.pulsar.binder.producerspring.cloud.stream.pulsar.binder.consumer

Pulsar 主题配置器

适用于 Apache Pulsar 的 Spring Cloud Stream 活页夹附带了用于 Pulsar 主题的开箱即用配置器。在运行应用程序时，如果缺少必要的主题，Pulsar 将为您创建主题。但是，这是一个基本的非分区主题，如果您想要创建分区主题等高级功能，则可以依赖活页夹中的主题配置程序。 Pulsar 主题配置器从框架中使用，该框架使用因此，除非您在默认服务器和端口上运行 Pulsar，否则您需要设置该属性。PulsarAdministrationPulsarAdminBuilder.spring.pulsar.administration.service-url

创建主题时指定分区计数

创建主题时，可以通过两种方式设置分区计数。首先，可以使用属性在活页夹级别设置它。正如我们在上面看到的，这样做将使应用程序创建的所有主题都继承此属性。假设您希望在绑定级别进行精细控制，以设置分区。在这种情况下，可以使用格式来设置每个绑定的属性。这样，同一应用程序中由不同函数创建的各种主题将根据应用程序需求具有不同的分区。spring.cloud.stream.pulsar.binder.partition-countpartition-countspring.cloud.stream.pulsar.bindings.<binding-name>.producer|consumer.partition-count