如何区分Kafka生产者发送的Protobuf与JSON消息并做针对性处理？

当然有可行方案，以下是几种实用的落地方式：

1. 消息头显式标记格式 #

这是最可靠的方案，要求生产者在发送消息时，给Kafka消息添加自定义Header字段（比如message-type），值设为protobuf或json。消费者端先读取这个Header，再根据标记选择对应的解析逻辑。

示例代码（Java）：

生产者添加Header #

ProducerRecord<String, byte[]> record = new ProducerRecord<>("topic", payload);
record.headers().add("message-type", "protobuf".getBytes(StandardCharsets.UTF_8));
producer.send(record);

消费者解析逻辑 #

ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, byte[]> record : records) {
    String msgType = new String(record.headers().lastHeader("message-type").value(), StandardCharsets.UTF_8);
    if ("protobuf".equals(msgType)) {
        // 用Protobuf解析payload
        MyProtoMsg msg = MyProtoMsg.parseFrom(record.value());
        // 处理逻辑
    } else if ("json".equals(msgType)) {
        // 用JSON解析payload
        ObjectMapper mapper = new ObjectMapper();
        MyJsonMsg msg = mapper.readValue(record.value(), MyJsonMsg.class);
        // 处理逻辑
    }
}

优点：识别准确率100%，逻辑清晰；缺点：需要修改生产者代码添加Header。

2. 基于消息内容的特征检测 #

如果无法修改生产者代码，可以通过消息内容的格式特征来判断：

• JSON是UTF-8编码的文本，开头通常是{或[，且能被JSON解析器正常解析；
• Protobuf是二进制格式，非UTF-8编码，且解析失败时会抛出特定异常。

消费者可以先尝试JSON解析，失败后再尝试Protobuf解析（或者反过来，根据业务场景调整顺序）。

示例代码（Python）：

import json
from my_proto_module import MyProtoMsg  # 导入Protobuf生成的类

def process_message(payload):
    try:
        # 先尝试解析JSON
        json_msg = json.loads(payload.decode('utf-8'))
        # JSON处理逻辑
        handle_json_message(json_msg)
    except (UnicodeDecodeError, json.JSONDecodeError):
        # 解码或JSON解析失败，尝试Protobuf
        try:
            proto_msg = MyProtoMsg.FromString(payload)
            # Protobuf处理逻辑
            handle_protobuf_message(proto_msg)
        except Exception as e:
            # 两种格式都解析失败，处理异常
            handle_invalid_message(payload, e)

优点：无需修改生产者；缺点：存在极小概率的误判（比如某些二进制刚好符合JSON开头），且解析失败会产生额外开销。

3. 按主题/分区拆分消息 #

如果业务允许调整，可以让生产者将Protobuf消息发送到专属主题（如order-events-protobuf），JSON消息发送到另一个主题（如order-events-json）。消费者直接订阅对应主题，分别处理不同格式的消息。

优点：完全避免格式判断的复杂度，逻辑最简洁；缺点：需要调整生产者的消息路由逻辑，可能涉及主题扩容。

4. 利用Schema Registry的多Schema支持 #

如果使用Confluent Schema Registry，可以为同一个主题注册多个Schema（Protobuf和JSON各一个）。生产者发送消息时指定对应的Schema ID，消费者通过消息中的schema-id字段获取对应的Schema，再进行解析。

示例逻辑：

• 生产者发送Protobuf消息时，关联Protobuf Schema的ID；发送JSON时关联JSON Schema的ID；
• 消费者从消息Header中读取schema-id，向Schema Registry获取对应Schema，再用该Schema解析消息。

优点：符合Schema管理规范，适合大规模生产环境；缺点：依赖Schema Registry组件，增加架构复杂度。

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