温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

怎么进行Pulsar Kafka Client的简单分析

发布时间:2021-12-15 09:19:02 来源:亿速云 阅读:143 作者:柒染 栏目:大数据

本篇文章给大家分享的是有关怎么进行Pulsar Kafka Client的简单分析,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说,跟着小编一起来看看吧。

为了方便 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@EqualsAndHashCodepublic class Foo {    @Nullable    private String field1;    @Nullable    private String field2;    private int field3;}
@Data@ToString@EqualsAndHashCodepublic 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的简单分析,小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI