溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇文章給大家分享的是有關怎么進行Pulsar Kafka Client的簡單分析，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。
?? 引入依賴
<dependency> <groupId>org.apache.pulsar</groupId> <artifactId>pulsar-client-kafka</artifactId> <version>{project.version}</version></dependency>
?? 使用 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();
>>> 添加消費者代碼
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 offsetconsumer.commitSync();}
?? 使用 Pulsar Schema
@Data@ToString@EqualsAndHashCodepublic class Foo {@Nullableprivate String field1;@Nullableprivate 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();
>>> 消費者端代碼
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 offsetconsumer.commitSync();}
以上就是怎么進行Pulsar Kafka Client的簡單分析，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
