消息制作
1. ReactivePulsarTemplate
在 Pulsar 生产者端,Spring Boot 自动配置提供了一个ReactivePulsarTemplate
用于发布记录。该模板实现了一个名为ReactivePulsarOperations
并提供通过其 Contract 发布记录的方法。
该模板提供了 send 方法,这些方法接受单个消息并返回Mono<MessageId>
.
它还提供了接受多条消息的 send 方法(以 ReactiveStreams 的形式Publisher
type) 并返回一个Flux<MessageId>
.
对于不包含 topic 参数的 API 变体,将使用主题解析过程来确定目标主题。 |
1.1. Fluent API
该模板提供了一个 Fluent 构建器来处理更复杂的发送请求。
1.2. 消息自定义
您可以指定MessageSpecBuilderCustomizer
以配置传出消息。例如,以下代码演示如何发送键控消息:
template.newMessage(msg)
.withMessageCustomizer((mc) -> mc.key("foo-msg-key"))
.send();
1.3. 发件人自定义
您可以指定ReactiveMessageSenderBuilderCustomizer
配置底层 Pulsar sender 构建器,最终构建用于发送传出消息的 sender。
请谨慎使用,因为这会提供对 sender 构建器的完全访问权限,并调用其某些方法(例如create ) 可能会产生意想不到的副作用。 |
例如,以下代码演示如何禁用批处理和启用分块:
template.newMessage(msg)
.withSenderCustomizer((sc) -> sc.enableChunking(true).enableBatching(false))
.send();
另一个示例显示了在将记录发布到分区主题时如何使用自定义路由。
指定您的自定义MessageRouter
在 Sender Builder 上实施,例如:
template.newMessage(msg)
.withSenderCustomizer((sc) -> sc.messageRouter(messageRouter))
.send();
请注意,当使用MessageRouter ,则spring.pulsar.producer.message-routing-mode 是custom . |
2. 指定 Schema 信息
如果您使用 Java 基元类型,框架会自动为您检测架构,并且您无需指定任何架构类型来发布数据。
对于非原始类型,如果在对ReactivePulsarTemplate
,Spring for Apache Pulsar 框架将尝试构建一个Schema.JSON
从 type.
当前支持的复杂架构类型包括 JSON、AVRO、PROTOBUF、AUTO_PRODUCE_BYTES 和 KEY_VALUE w/ INLINE 编码。 |
2.1. 自定义 Schema 映射
作为在ReactivePulsarTemplate
对于复杂类型,可以使用类型的映射配置 Schema Resolver。
这样就无需在框架使用传出消息类型咨询解析程序时指定架构。
2.1.1. 配置属性
架构映射可以使用spring.pulsar.defaults.type-mappings
财产。
以下示例使用application.yml
要为User
和Address
复杂对象 usingAVRO
和JSON
schemas 的 Schemas:
spring:
pulsar:
defaults:
type-mappings:
- message-type: com.acme.User
schema-info:
schema-type: AVRO
- message-type: com.acme.Address
schema-info:
schema-type: JSON
这message-type 是 Message 类的完全限定名称。 |
2.1.2. Schema 解析器定制器
添加映射的首选方法是通过上述属性。 但是,如果需要更多控制,则可以提供架构解析程序定制器来添加映射。
以下示例使用架构解析程序定制器为User
和Address
复杂对象 usingAVRO
和JSON
schemas 的 Schemas:
@Bean
public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (schemaResolver) -> {
schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class));
}
}
2.2. 使用 AUTO_SCHEMA 进行生产
如果无法提前知道 Pulsar topic 的 schema 类型,可以使用 AUTO_PRODUCE schema 将原始 JSON 或 Avro payload 发布为byte[]
牢牢。
在这种情况下,创建者会验证出站字节是否与目标主题的架构兼容。
只需指定一个Schema.AUTO_PRODUCE_BYTES()
在您的模板 send作中,如下例所示:
void sendUserAsBytes(ReactivePulsarTemplate<byte[]> template, byte[] userAsBytes) {
template.send("user-topic", userAsBytes, Schema.AUTO_PRODUCE_BYTES()).subscribe();
}
这仅支持 Avro 和 JSON 架构类型。 |
3. 响应式 PulsarSenderFactory
这ReactivePulsarTemplate
依赖于ReactivePulsarSenderFactory
以实际创建基础发件人。
Spring Boot 提供了这个发送者工厂,可以使用任何spring.pulsar.producer.*
应用程序属性。
3.1. 生产者缓存
每个底层 Pulsar 生产者都会消耗资源。
为了提高性能并避免持续创建 producer,ReactiveMessageSenderCache
在底层 Apache Pulsar Reactive 客户端缓存它创建的生产者。
它们以 LRU 方式缓存,并在配置的时间段内未使用时被驱逐。
您可以通过指定任何spring.pulsar.producer.cache.*
应用程序属性。