kafka传输自定义对象

在 Kafka 中传输自定义对象，需要进行对象的序列化和反序列化。以下是一个简单的示例，演示如何使用 Kafka 和 Avro 进行自定义对象的传输。

首先，您需要创建一个用于序列化和反序列化对象的 Avro schema，可以使用 Avro 的 IDL 或者 JSON 格式定义。假设我们有一个名为 User 的自定义对象，具有字段 name 和 age：

{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}

然后，您需要为该 schema 生成对应的 Java 类。可以使用 Avro 的工具将 schema 文件转换为 Java 类。假设生成的类名为 User。

接下来，您需要配置 Kafka 的生产者和消费者，以便使用 Avro 进行序列化和反序列化。在生产者端，您需要指定 Avro 的序列化器，并将 Avro 类型的对象作为消息发送到 Kafka：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

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

User user = new User();
user.setName("John");
user.setAge(30);

ProducerRecord<String, User> record = new ProducerRecord<>("topic-name", "key", user);
producer.send(record);

在消费者端，您需要指定 Avro 的反序列化器，并接收 Avro 类型的对象：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", KafkaAvroDeserializer.class);
props.put("schema.registry.url", "http://localhost:8081");

Consumer<String, User> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic-name"));

ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, User> record : records) {
  String key = record.key();
  User user = record.value();
  System.out.println("Key: " + key + ", User: " + user.getName() + ", " + user.getAge());
}

确保在您的项目中添加以下依赖项：

<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
  </dependency>
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>6.2.0</version>
  </dependency>
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry</artifactId>
    <version>6.2.0</version>
  </dependency>
</dependencies>

请注意，上述示例假设您已经正确设置了 Kafka 和 Schema Registry，并且可以连接到这些服务。还要确保 Avro 的依赖项与 Kafka 版本兼容。

这只是一个简单的示例，您可以根据自己的需求进行相应的调整和扩展。希望对您有所帮助！