Pulsar Kafka Client 简单介绍

🎙️阅读本文需要 5 分钟

为了方便 Kafka 用户使用 Pulsar，Pulsar 对 Kafka Client 做了一些封装，让 Kafka 用户更方便的使用 Pulsar。

本篇内容主要介绍 Kafka Client 如何将消息发送到 Pulsar, 并从 Pulsar 消费消息，以及如何使用 Pulsar Schema。

---

⌨️ 引入依赖

<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client-kafka</artifactId>
  <version>{project.version}</version>
</dependency>

依赖引入了 Kafka 的 0.10.2.1 版本的客户端，还有 Pulsar 对 Kafka Client 封装后的客户端。

⌨️ 使用 Kafka Schema

>>> 添加生产者代码

String topic = "persistent://public/default/test";

Properties props = new Properties();
props.put("bootstrap.servers", "pulsar://localhost:6650");

props.put("key.serializer", IntegerSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());

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

for (int i = 0; i < 10; i++) {
    producer.send(new ProducerRecord<Integer, String>(topic, i, Integer.toString(i)));
}

producer.close();

在上述配置中 topic 是指 Pulsar 中的 Topic，接着使用 Kafka 的配置方式来初始化各种配置，包括 Server 地址、key 的序列化与 value 的序列化类，然后构造一个 ProducerRecord 的类将其发送出去。

>>> 添加消费者代码

String topic = "persistent://public/default/test";

Properties props = new Properties();
props.put("bootstrap.servers", "pulsar://localhost:6650");
props.put("group.id", "my-subscription-name");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", IntegerDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

@SuppressWarnings("resource")
Consumer<Integer, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));

while (true) {
    ConsumerRecords<Integer, String> records = consumer.poll(100);
    records.forEach(record -> {
        log.info("Received record: {}", record);
    });

    // Commit last offset
    consumer.commitSync();
}

有些配置同生产者代码的配置是类似的，例如 topic，Server 等。另外使用 Kafka 的 group.id 作为配置 Pulsar 中的订阅名称，关闭自动提交，在消费者端为 key 和 value 配置的是反序列化的类。然后同常规的消费者类似，开始消费消息。

---

⌨️ 使用 Pulsar Schema

在上述情况中使用的是 Kafka 的 Schema 来进行序列化与反序列化，当然也支持使用 Pulsar 的 Schema 来进行此过程。下面使用 AVRO 进行简单的介绍。

首先定义 Schema 所需要使用的 pojo 类。

@Data
@ToString
@EqualsAndHashCode
public class Foo {
    @Nullable
    private String field1;
    @Nullable
    private String field2;
    private int field3;
}

@Data
@ToString
@EqualsAndHashCode
public class Bar {
    private boolean field1;
}

>>> 生产者端代码

String topic = "persistent://public/default/test-avro";

Properties props = new Properties();
props.put("bootstrap.servers", "pulsar://localhost:6650");

props.put("key.serializer", IntegerSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());

AvroSchema<Bar> barSchema = AvroSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build());
AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build());

Bar bar = new Bar();
bar.setField1(true);

Foo foo = new Foo();
foo.setField1("field1");
foo.setField2("field2");
foo.setField3(3);


Producer<Foo, Bar> producer = new KafkaProducer<>(props, fooSchema, barSchema);

for (int i = 0; i < 10; i++) {
    producer.send(new ProducerRecord<Foo, Bar>(topic, i, foo, bar));
    log.info("Message {} sent successfully", i);
}

producer.close();

可以看到大部分配置同上面使用 Kafka Client 的配置是类似的，但是中间加入了一些 Pulsar 的 Schema，使用 Foo 作为 key，使用 Bar 类作为 value。

>>> 消费者端代码

String topic = "persistent://public/default/test-avro";

Properties props = new Properties();
props.put("bootstrap.servers", "pulsar://localhost:6650");
props.put("group.id", "my-subscription-name");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", IntegerDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

AvroSchema<Bar> barSchema = AvroSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build());
AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build());

Bar bar = new Bar();
bar.setField1(true);

Foo foo = new Foo();
foo.setField1("field1");
foo.setField2("field2");
foo.setField3(3);

@SuppressWarnings("resource")
Consumer<Foo, Bar> consumer = new PulsarKafkaConsumer<>(props, fooSchema, barSchema);
consumer.subscribe(Arrays.asList(topic));

while (true) {
    ConsumerRecords<Foo, Bar> records = consumer.poll(100);
    records.forEach(record -> {
        log.info("Received record: {}", record);
    });

    // Commit last offset
    consumer.commitSync();
}

消费者端同样是类似的配置，使用与生产者端相同的 Schema 进行数据的反序列化。

---

📣 总结一下

本文简单介绍了一些关于在 Pulsar 中使用 Kafka Client 的一些内容，pulsar-kafka-client 做了封装，方便用户使用 Kafka Client 与 Pulsar 打交道。

更多关于 Kafka Client 中支持的配置，可以参考 Pulsar 官网。

🔗http://pulsar.apache.org/docs/en/adaptors-kafka/#compatibility-matrix

点击「阅读原文」，查看 Kafka Client 支持的配置。