MassTransit Kafka空消息处理：如何消费值为空但Key非空的消息？

完全可以消费这类消息，默认情况下被跳过通常是因为客户端反序列化器对null值的处理逻辑导致的，调整配置或代码即可解决：

核心原因 #

Kafka本身允许消息value为null、key非空的合法记录，多数客户端（如Java KafkaConsumer）的默认反序列化器在遇到value为null时，可能抛出异常或直接过滤掉这条消息，导致业务代码无法感知到它。

解决方法 #

1. 自定义支持null值的反序列化器 #

针对你使用的客户端语言，实现一个能处理null value的反序列化器：
以Java为例，自定义String类型的反序列化器：

import org.apache.kafka.common.serialization.Deserializer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class NullSafeStringDeserializer implements Deserializer<String> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public String deserialize(String topic, byte[] data) {
        return data == null ? null : new String(data, StandardCharsets.UTF_8);
    }

    @Override
    public void close() {}
}

然后在消费者配置中指定这个反序列化器：

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NullSafeStringDeserializer.class);

2. 直接使用原始消费API处理 #

在消费循环中，直接判断ConsumerRecord的value是否为null，无需依赖反序列化器的特殊处理：
Java示例：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        if (record.value() == null) {
            // 处理value为null、key非空的消息
            System.out.printf("处理消息：topic=%s, key=%s, value=null%n", record.topic(), record.key());
        } else {
            // 正常消息处理逻辑
        }
    }
}

3. 其他客户端的注意事项 #

比如Python的kafka-python库，默认就支持接收value为None的消息，只需在消费时添加判断逻辑即可，无需额外配置。

内容的提问来源于stack exchange，提问作者Isard