空载荷和“墓碑”记录的日志压缩
使用日志压缩时,您可以发送和接收带有 null
载荷的消息来标识键的删除。您也可能由于其他原因接收到 null
值,例如反序列化器在无法反序列化值时可能返回 null
。
生产空载荷
您可以通过将 null
消息参数值传递给 ReactivePulsarTemplate
的其中一个 send
方法来发送 null
值,例如
reactiveTemplate
.send(null, Schema.STRING)
.subscribe();
发送 null 值时,必须指定 Schema 类型,因为系统无法从 null 载荷确定消息的类型。 |
消费空载荷
对于 `@ReactivePulsarListener`,`null` 载荷根据其消息参数的类型传递到监听器方法中,具体如下
参数类型 | 传入的值 |
---|---|
基本类型 |
|
用户定义类型 |
|
|
一个非空的 Pulsar 消息,其 |
|
一个非空的 Spring 消息,其 |
|
一个非空的 Flux,其条目是非空的 Pulsar 消息,这些消息的 |
|
一个非空的 Flux,其条目是非空的 Spring 消息,这些消息的 |
当传入的值为 null 时(即使用基本类型或用户定义类型的单记录监听器),您必须使用 `@Payload` 参数注解并设置 required = false 。 |
当使用 Spring 的 org.springframework.messaging.Message 作为监听器的载荷类型时,其泛型类型信息必须足够宽泛,以接受 Message<PulsarNull> (例如 Message 、Message<?> 或 Message<Object> )。这是因为 Spring Message 不允许其载荷为 null 值,而是使用 PulsarNull 占位符。 |
如果它是用于压缩日志的墓碑消息,您通常还需要获取 key,以便您的应用程序可以确定哪个 key 被“`删除`”了。以下示例展示了这种配置
@ReactivePulsarListener(
topics = "my-topic",
subscriptionName = "my-topic-sub",
schemaType = SchemaType.STRING)
Mono<Void> myListener(
@Payload(required = false) String msg,
@Header(PulsarHeaders.KEY) String key) {
...
}
使用流式消息监听器(Flux )时,头部支持有限,因此在日志压缩场景中不太有用。 |