空负载和“墓碑”记录的对数压缩

使用日志压缩时,您可以发送和接收带有null负载的消息来标识键的删除。您也可能由于其他原因收到null值,例如反序列化器在无法反序列化值时可能会返回null

生成空负载

您可以通过将null消息参数值传递给send方法之一,例如,使用ReactivePulsarTemplate发送null

reactiveTemplate
        .send(null, Schema.STRING)
        .subscribe();
发送空值时,必须指定模式类型,因为系统无法从null有效负载中确定消息类型。

消费空有效负载

对于@ReactivePularListenernull有效负载将根据其消息参数的类型传递到监听器方法中,如下所示

参数类型 传入值

基本类型

null

用户定义

null

org.apache.pulsar.client.api.Message<T>

非空 Pulsar 消息,其getValue()返回null

org.springframework.messaging.Message<T>

非空 Spring 消息,其getPayload()返回PulsarNull

Flux<org.apache.pulsar.client.api.Message<T>>

非空流,其条目是非空 Pulsar 消息,其getValue()返回null

Flux<org.springframework.messaging.Message<T>>

非空流,其条目是非空 Spring 消息,其getPayload()返回PulsarNull

当传入值为null(即具有基本类型或用户定义类型的单个记录监听器)时,必须使用@Payload参数注释,其中required = false
当使用 Spring org.springframework.messaging.Message 作为监听器有效负载类型时,其泛型类型信息必须足够宽泛以接受Message<PulsarNull>(例如,MessageMessage<?>Message<Object>)。这是因为 Spring 消息不允许其有效负载为空值,而是使用PulsarNull占位符。

如果是压缩日志的墓碑消息,通常还需要键,以便应用程序可以确定哪个键被“删除”。以下示例显示了这种配置

@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)时,头支持有限,因此在日志压缩场景中不太有用。