Kafka是否使用AVRO、JSONSchema或Protobuf对消息进行二进制存储？

Kafka 可以使用任何序列化格式对消息进行二进制存储。其中 AVRO、JSON Schema 和 Protobuf 都是常见的选项。

以下是使用 AVRO 作为序列化格式的示例代码：

1. 添加 Maven 依赖：

<dependency>
   <groupId>io.confluent</groupId>
   <artifactId>kafka-schema-registry-client</artifactId>
   <version>5.4.1</version>
</dependency>
<dependency>
   <groupId>io.confluent</groupId>
   <artifactId>kafka-avro-serializer</artifactId>
   <version>5.4.1</version>
</dependency>

2. 配置 Kafka 生产者：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

3. 定义 AVRO 模式：

{
   "namespace": "example.avro",
   "type": "record",
   "name": "User",
   "fields": [
      {"name": "name", "type": "string"},
      {"name": "favorite_number",  "type": ["int", "null"]},
      {"name": "favorite_color", "type": ["string", "null"]}
   ]
}

4. 创建一个 AVRO 消息：

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(new File("src/main/avro/user.avsc"));
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("name", "Alice");
avroRecord.put("favorite_number", 42);
avroRecord.put("favorite_color", "purple");

5. 发送消息到 Kafka：

Producer<String, GenericRecord> producer = new KafkaProducer<>(props);
ProducerRecord<String