Spring for Apache Pulsar->Reactive Support->Message Consumption
1. @ReactivePulsarListener
对于Pulsar消费者,我们建议最终用户应用程序使用ReactivePulsarListener注释。要使用ReactivePulsarListener,您需要使用@EnableReactivePulse注释。当您使用Spring Boot支持时,它会自动启用此注释并配置所有必要的组件,例如消息侦听器基础设施(负责创建底层Pulsar消费者)。
让我们重新审视我们在快速浏览部分看到的ReactivePulse Listener代码片段:
@ReactivePulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
Mono<Void> listen(String message) {System.out.println(message);return Mono.empty();
}
监听器方法返回一个Mono<Void>,以表示消息是否已成功处理。Mono.empty()表示成功(确认),Mono.error()表示失败(否定确认)。
您还可以进一步简化此方法:
@ReactivePulsarListener
Mono<Void> listen(String message) {System.out.println(message);return Mono.empty();
}
在这种最基本的形式中,当没有直接提供主题时,使用主题解析过程来确定目标主题。同样,当@ReactivePulsarListener注释上未提供subscriptionName时,将使用自动生成的订阅名称。
在前面显示的ReactivePulse Listener方法中,我们以String形式接收数据,但我们没有指定任何模式类型。在内部,该框架依赖Pulsar的模式机制将数据转换为所需的类型。
框架检测到您期望的String类型,然后根据该信息推断模式类型,并将该模式提供给消费者。该框架对所有原始类型都进行了这种推理。对于所有非基元类型,默认模式假定为JSON。如果复杂类型使用JSON以外的任何内容(如AVRO或KEY_VALUE),则必须使用schemaType属性在注释上提供模式类型。
此示例显示了如何从主题中使用复杂类型:
@ReactivePulsarListener(topics = "my-topic-2", schemaType = SchemaType.JSON)
Mono<Void> listen(Foo message) {System.out.println(message);return Mono.empty();
}
让我们再看看我们可以消费的几种方式。
此示例直接使用Pulsar消息:
@ReactivePulsarListener(topics = "my-topic")
Mono<Void> listen(org.apache.pulsar.client.api.Message<String> message) {System.out.println(message.getValue());return Mono.empty();
}
此示例使用包装在Spring消息信封中的记录:
@ReactivePulsarListener(topics = "my-topic")
Mono<Void> listen(org.springframework.messaging.Message<String> message) {System.out.println(message.getPayload());return Mono.empty();
}
1.1. Streaming
以上所有内容都是逐一消费单个记录的示例。然而,使用Reactive的一个令人信服的原因是具有背压支持的流媒体功能。
以下示例使用ReactivePulsarListener来消耗POJO流:
@ReactivePulsarListener(topics = "streaming-1", stream = true)
Flux<MessageResult<Void>> listen(Flux<org.apache.pulsar.client.api.Message<String>> messages) {return messages.doOnNext((msg) -> System.out.println("Received: " + msg.getValue())).map(MessageResult::acknowledge);
}
在这里,我们以脉冲星消息流的形式接收记录。此外,要在ReactivePulse Listener级别启用流消费,您需要将注释上的流属性设置为true。
监听器方法返回一个Flux<MessageResult<Void>>,其中每个元素表示一个已处理的消息,并保存消息id、值以及是否被确认。MessageResult有一组静态工厂方法,可用于创建相应的MessageResult实例。
根据Flux中消息的实际类型,框架试图推断要使用的模式。如果它包含复杂类型,您仍然需要在ReactivePulsarListener上提供schemaType。
以下侦听器使用具有复杂类型的Spring消息传递消息信封:
@ReactivePulsarListener(topics = "streaming-2", stream = true, schemaType = SchemaType.JSON)
Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo>> messages) {return messages.doOnNext((msg) -> System.out.println("Received: " + msg.getPayload())).map(MessageUtils::acknowledge);
}
监听器方法返回一个Flux<MessageResult<Void>>,其中每个元素表示一个已处理的消息,并保存消息id、值以及是否被确认。SpringMessageUtils有一组静态工厂方法,可用于从Spring消息创建相应的MessageResult实例。MessageUtils为Spring消息提供了与MessagResult上的工厂方法集为Pulsar消息提供的功能相同的功能。
不支持使用org.apache.pulser.client.api。@ReactivePulsarListener中的消息<T>
1.2. Configuration - Application Properties
监听器依赖ReactivePulsarConsumerFactory来创建和管理它用来消费消息的底层Pulsar消费者。Spring Boot提供了这个消费者工厂,您可以通过指定Spring.pulser.consumer.*应用程序属性来进一步配置它。
1.3. Generic records with AUTO_CONSUME
如果没有机会提前知道Pulsar主题的模式类型,您可以使用AUTO_CONSUME模式类型来使用通用记录。在这种情况下,主题使用与主题关联的模式信息将消息反序列化为GenericRecord对象。
要使用通用记录,请设置schemaType=schemaType。在@ReactivePulsarListener上设置AUTO_CONSUME,并使用GenericRecord类型的Pulsar消息作为消息参数,如下所示。
@ReactivePulsarListener(topics = "my-generic-topic", schemaType = SchemaType.AUTO_CONSUME)
Mono<Void> listen(org.apache.pulsar.client.api.Message<GenericRecord> message) {GenericRecord record = message.getValue();record.getFields().forEach((f) ->System.out.printf("%s = %s%n", f.getName(), record.getField(f)));return Mono.empty();
}
GenericRecord API允许访问字段及其关联值
1.4. Consumer Customization
您可以指定ReactivePulsarListenerMessageConsumerBuilderCustomizer来配置底层Pulsar消费者构建器,该构建器最终构建了监听器用于接收消息的消费者。
请谨慎使用,因为这可以完全访问消费者构建器,调用其某些方法(如create)可能会产生意想不到的副作用。
例如,以下代码显示了如何将订阅的初始位置设置为主题上最早的消息。
@ReactivePulsarListener(topics = "hello-pulsar-topic", consumerCustomizer = "myConsumerCustomizer")
Mono<Void> listen(String message) {System.out.println(message);return Mono.empty();
}@Bean
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
}
如果您的应用程序只注册了一个@ReactivePulsarListener和一个ReactivePulse ListenerMessageConsumerBuilderCustomizer bean,则会自动应用自定义程序。
您还可以使用定制器向消费者构建器提供Pulsar消费者的直接属性。如果您不想使用前面提到的Boot配置属性,或者有多个配置不同的ReactivePulse Listener方法,这很方便。
以下自定义程序示例使用Pulsar直接消费者属性:
@Bean
ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1");
}
The properties used are direct Pulsar consumer properties, not the spring.pulsar.consumer Spring Boot configuration properties |
2. Specifying Schema Information
如前所述,对于Java原语,Spring for Apache Pulsar框架可以推断出在ReactivePulsarListener上使用的正确模式。对于非基元类型,如果没有在注释上明确指定Schema,Spring For Apache Pulsar框架将尝试构建Schema。JSON类型。
目前支持的复杂模式类型有JSON、AVRO、PROTOBUF、AUTO_CONSUME、KEY_VALUE和INLINE编码。
2.1. Custom Schema Mapping
作为在ReactivePulse Listener上为复杂类型指定模式的替代方案,可以使用类型的映射配置模式解析器。这消除了在侦听器上设置模式的需要,因为框架使用传入消息类型咨询解析器。
2.1.1. Configuration properties
模式映射可以使用spring.pulsar.defaults.type-mappings属性进行配置。以下示例使用application.yml分别使用AVRO和JSON模式为User和Address复杂对象添加映射:
spring:pulsar:defaults:type-mappings:- message-type: com.acme.Userschema-info:schema-type: AVRO- message-type: com.acme.Addressschema-info:schema-type: JSON
消息类型是消息类的完全限定名。
2.1.2. Schema resolver customizer
添加映射的首选方法是通过上述属性。但是,如果需要更多的控制,您可以提供一个模式解析器定制器来添加映射。
以下示例使用模式解析器定制器分别使用AVRO和JSON模式为User和Address复杂对象添加映射:
@Bean
public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {return (schemaResolver) -> {schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class));}
}
2.1.3. Type mapping annotation
指定用于特定消息类型的默认模式信息的另一种选择是用@PulsarMessage注释标记消息类。可以通过注释上的schemaType属性指定架构信息。
以下示例将系统配置为在生成或使用Foo类型的消息时使用JSON作为默认模式:
@PulsarMessage(schemaType = SchemaType.JSON)
record Foo(String value) {
}
有了这个配置,就不需要在监听器上设置模式,例如:
@ReactivePulsarListener(topics = "user-topic")
Mono<Void> listen(User user) {System.out.println(user);return Mono.empty();
}
3. Message Listener Container Infrastructure
在大多数情况下,我们建议直接使用ReactivePulsarListener注释从Pulsar主题中消费,因为该模型涵盖了广泛的应用程序用例。然而,了解ReactivePulsarListener的内部工作原理非常重要。
当您使用Spring for Apache Pulsar时,消息监听器容器是消息消费的核心。ReactivePulsarListener在幕后使用消息侦听器容器基础设施来创建和管理底层Pulsar消费者。
3.1. ReactivePulsarMessageListenerContainer
此消息侦听器容器的合约是通过ReactivePulsarMessageListenerContainer提供的,其默认实现创建了一个反应式Pulsar消费者,并连接了一个使用创建的消费者的反应式消息管道。
3.2. ReactiveMessagePipeline
管道是底层Apache Pulsar响应式客户端的一个功能,它以响应式方式接收数据,然后将其移交给提供的消息处理程序。反应式消息侦听器容器实现要简单得多,因为管道处理了大部分工作。
3.3. ReactivePulsarMessageHandler
“监听器”方面由ReactivePulse MessageHandler提供,它提供了两种实现:
ReactivePulsarOneByOneMessageHandler-逐一处理单个消息
ReactivePulsarStreamingHandler-通过Flux处理多条消息
如果在直接使用侦听器容器时未指定主题信息,则使用ReactivePulse listener使用的相同主题解析过程,但省略了“消息类型默认”步骤。
3.4. Handling Startup Failures
消息侦听器容器在刷新应用程序上下文时启动。默认情况下,启动过程中遇到的任何故障都会被重新抛出,应用程序将无法启动。您可以使用相应容器属性上的StartupFailurePolicy调整此行为。
可用选项包括:
Stop(默认)-记录并重新抛出异常,有效地停止应用程序
继续-记录异常,使容器处于非运行状态,但不要停止应用程序
重试-记录异常,重试异步启动容器,但不要停止应用程序。
默认的重试行为是重试3次,每次尝试之间有10秒的延迟。但是,可以在相应的容器属性上指定自定义重试模板。如果容器在重试尝试结束后未能重新启动,它将处于非运行状态。
3.4.1. Configuration
With Spring Boot
使用Spring Boot时,您可以注册PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>>bean,用于设置容器启动属性。
Without Spring Boot
但是,如果您改为手动配置组件,则在构建消息侦听器容器工厂时,必须相应地更新容器启动属性。
4. Concurrency
当以流模式(stream=true)消费记录时,并发性自然会通过客户端实现中的底层响应式支持来实现。
然而,在逐一处理消息时,可以指定并发性以提高处理吞吐量。只需在@ReactivePulsarListener上设置并发属性。此外,当并发性>1时,您可以通过在注释上设置useKeyOrderedProcessing=“true”来确保消息按键排序,从而发送到同一个处理程序。
同样,ReactiveMessagePipeline完成了繁重的工作,我们只需设置它的属性。
反应式容器中的并发性不同于其命令式对应物。后者创建多个线程(每个线程都有一个Pulsar消费者),而前者在响应式并行调度器上将消息并发分派给多个处理程序实例。
反应式并发模型的一个优点是,它可以与独占订阅一起使用,而命令式并发模型则不能。
5. Pulsar Headers
Pulsar消息元数据可以作为Spring消息头使用。可用标头的列表可以在PulsarHeaders.java中找到。
5.1. Accessing In OneByOne Listener
以下示例显示了使用逐一消息侦听器时如何访问Pulsar Header:
@ReactivePulsarListener(topics = "some-topic")
Mono<Void> listen(String data,@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,@Header("foo") String foo) {System.out.println("Received " + data + " w/ id=" + messageId + " w/ foo=" + foo);return Mono.empty();
}
在前面的示例中,我们访问messageId消息元数据的值以及名为foo的自定义消息属性。Spring@Header注释用于每个标头字段。
您还可以使用Pulsar的消息作为信封来携带有效载荷。这样做时,用户可以直接调用Pulsar消息上的相应方法来检索元数据。但是,为了方便起见,您还可以使用Header注释来检索它。请注意,您还可以使用Spring消息传递消息信封来携带有效载荷,然后使用@Header检索Pulsar标头。
5.2. Accessing In Streaming Listener
使用流式消息侦听器时,标头支持是有限的。仅当Flux包含Spring org.springframework.messing时。消息元素将填充标头。此外,Spring@Header注释不能用于检索数据。您必须直接调用Spring消息上的相应方法来检索数据。
6. Message Acknowledgment
该框架自动处理消息确认。但是,侦听器方法必须发送一个信号,指示消息是否已成功处理。然后,容器实现使用该信号执行ack或nack操作。这与它的命令式对应略有不同,在命令式对应中,除非方法抛出异常,否则信号被暗示为正。
6.1. OneByOne Listener
单消息(又名OneByOne)消息侦听器方法返回Mono<Void>,以表示消息是否已成功处理。Mono.empty()表示成功(确认),Mono.error()表示失败(否定确认)。
6.2. Streaming Listener
流式监听器方法返回一个Flux<MessageResult<Void>>,其中每个MessageResult元素表示一个已处理的消息,并保存消息id、值以及是否已确认。MessageResult有一组acknowledge和negativeAcknowledge静态工厂方法,可用于创建相应的MessageResult实例。
7. Message Redelivery and Error Handling
Apache Pulsar为消息重新传递和错误处理提供了各种本机策略。我们将看看它们,看看如何在Spring for Apache Pulsar中使用它们。
7.1. Acknowledgment Timeout
默认情况下,Pulsar消费者不会重新传递消息,除非消费者崩溃,但您可以通过在Pulsar消费者上设置ack超时来更改此行为。如果ack timeout属性的值大于零,并且Pulsar消费者在该超时时间内未确认消息,则重新传递消息。
您可以通过消费者定制器直接将此属性指定为Pulsar消费者属性,例如:
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {return b -> b.property("ackTimeoutMillis", "60000");
}
7.2. Negative Acknowledgment Redelivery Delay
当确认为否定时,Pulsar消费者允许您指定应用程序希望如何重新传递消息。默认情况是在一分钟内重新传递消息,但您可以通过消费者自定义程序进行更改,例如:
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {return b -> b.property("negativeAckRedeliveryDelay", "10ms");
}
7.3. Dead Letter Topic
Apache Pulsar允许应用程序在共享订阅类型的消费者身上使用死信主题。对于独占和故障转移订阅类型,此功能不可用。基本思想是,如果消息被重试了一定次数(可能是由于ack超时或nack重新传递),一旦重试次数用尽,消息就可以被发送到一个称为死信队列(DLQ)的特殊主题。让我们通过检查一些代码片段来了解此功能的一些细节:
@Configuration(proxyBeanMethods = false)
class DeadLetterPolicyConfig {@ReactivePulsarListener(topics = "topic-with-dlp",subscriptionType = SubscriptionType.Shared,deadLetterPolicy = "myDeadLetterPolicy",consumerCustomizer = "ackTimeoutCustomizer" )void listen(String msg) {throw new RuntimeException("fail " + msg);}@ReactivePulsarListener(topics = "my-dlq-topic")void listenDlq(String msg) {System.out.println("From DLQ: " + msg);}@BeanDeadLetterPolicy myDeadLetterPolicy() {return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build();}@BeanReactiveMessageConsumerBuilderCustomizer<String> ackTimeoutCustomizer() {return b -> b.property("ackTimeoutMillis", "1000");}
}
首先,我们为DeadLetterPolicy提供了一个特殊的bean,它被命名为DeadLetterPolicy(它可以是您想要的任何名称)。此bean指定了许多内容,例如最大传递率(在本例中为10)和死信主题的名称 — 在这种情况下,我的dlq主题。如果不指定DLQ主题名称,Pulsar中默认为<topicname>-<subscriptionname>-DLQ。接下来,我们通过设置deadLetterPolicy属性将此bean名称提供给ReactivePulsarListener。请注意,ReactivePulse Listener的订阅类型为共享,因为DLQ功能仅适用于共享订阅。此代码主要用于演示目的,因此我们提供了ackTimeoutMillis值1000。这个想法是,代码抛出异常,如果Pulsar在1秒内没有收到ack,它会重试。如果该循环持续十次(因为这是我们在DeadLetterPolicy中的最大重新交付计数),Pulsar消费者会将消息发布到DLQ主题。我们还有另一个ReactivePulse Listener,它监听DLQ主题,以接收发布到DLQ主题的数据。
如果主主题在幕后被分区,Pulsar会将每个分区视为单独的主题。Pulsar在主主题名称后附加partition-<n>,其中n代表分区号。问题是,如果你不指定DLQ主题(与我们上面所做的相反),Pulsar会发布一个默认的主题名称,其中包含“分区-<n>信息” — 例如:topic-with-dlp-partition-0-deadLetterPolicy订阅-DLQ。解决这个问题的简单方法是始终提供DLQ主题名称。
8. Pulsar Reader Support
该框架支持通过ReactivePulse ReaderFactory以响应式方式使用Pulsar Reader。
Spring Boot提供了这个阅读器工厂,可以配置任何Spring.pulser.reader.*应用程序属性。