Null Payloads and Log Compaction of 'Tombstone' Records

在使用日志压缩时，您可以发送和接收具有 null 有效负载的消息，以识别密钥的删除。您还可以出于其他原因接收 null 值，例如反序列化程序在无法反序列化值时可能会返回 null。

Producing Null Payloads

您可以通过将 null 消息参数值传递到 send 方法之一，以带 ReactivePulsarTemplate 发送一个 null 值，例如：

reactiveTemplate
        .send(null, Schema.STRING)
        .subscribe();

发送空值时，您必须指定模式类型，因为系统无法从 null 有效负载确定消息的类型。

Consuming Null Payloads

对于 @ReactivePularListener，根据其消息参数类型将 null 载荷传递到侦听器方法，如下所示：

Parameter type Passed-in value

primitive

null

user-defined

null

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

非 null Pulsar 消息，其 getValue() 返回 null

org.springframework.messaging.Message<T>

非 null Spring 消息，其 getPayload() 返回 PulsarNull

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

非空流，其项为非空 Pulsar 消息，它的 getValue() 返回 null

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

非空流，其项为非空 Spring 消息，它的 getPayload() 返回 PulsarNull

当传入参数为 null（例如，带有原始类型或用户自定义类型的单记录侦听器）时，您必须使用 @Payload 带有 required = false 的参数注释。

当为您的侦听器负载类型使用 Spring org.springframework.messaging.Message 时，其泛型类型信息必须足够广泛，以接受 Message<PulsarNull>（例如 Message、Message<?> 或 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 ）时，header support is limited，因此在日志压实场景中不太有用。