1.概述

请参考：https://www.jianshu.com/p/a70950bab06d

【Kafka】Kafka 使用传统的 avro API 自定义序列化类和反序列化类比较麻烦，需要根据 schema 生成实体类，需要调用 avro 的 API 实现对象到 byte[] 和 byte[] 到对象的转化，而那些方法看上去比较繁琐，幸运的是，Twitter 开源的类库 Bijection 对传统的 Avro API 进行了封装了和优化，让我们可以方便的实现以上操作。

2.依赖

Bijection 类库的依赖如下：

<dependency><groupId>com.twitter</groupId><artifactId>bijection-avro_2.11</artifactId><version>0.9.6</version>
</dependency>

在 maven 工程的 resources 目录下新建一个 schema 文件，名称为"user.json"，因为我们不用 avro 生成实体类的方式，所以定义一个普通的 json 文件来描述 schema 即可，另外，在 json 文件中，也不需要"namespace": "packageName"这个限定生成实体类的包名的参数，本文使用的 json 文件内容如下：

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

3.测试类

package com.kafka.avro.demo.bijection;import static org.junit.Assert.*;import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Collections;
import java.util.Properties;import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;import com.twitter.bijection.Injection;
import com.twitter.bijection.avro.GenericAvroCodecs;
import org.junit.Test;public class BijectionProducerTest {@Testpublic void producerTest() throws Exception {String schemaFilePath = BijectionProducer.class.getClassLoader().getResource("user.json").getPath();FileReader fr = new FileReader(new File(schemaFilePath));BufferedReader br = new BufferedReader(fr);StringBuilder sb = new StringBuilder();String line;while ((line = br.readLine()) != null) {sb.append(line).append("\n");}String schemaStr = sb.toString();br.close();fr.close();Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");Schema.Parser parser = new Schema.Parser();Schema schema = parser.parse(schemaStr);Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);Producer<String, byte[]> producer = new KafkaProducer<>(props);for (int i = 0; i < 100; i++) {GenericData.Record avroRecord = new GenericData.Record(schema);avroRecord.put("id", i);avroRecord.put("name", "name" + i);avroRecord.put("age", 22);byte[] avroRecordBytes = recordInjection.apply(avroRecord);ProducerRecord<String, byte[]> record = new ProducerRecord<>("topic_lcc", avroRecordBytes);producer.send(record);Thread.sleep(1000);}producer.close();}@Testpublic void consumerTest() throws Exception {String schemaFilePath = BijectionProducer.class.getClassLoader().getResource("user.json").getPath();FileReader fr = new FileReader(new File(schemaFilePath));BufferedReader br = new BufferedReader(fr);StringBuilder sb = new StringBuilder();String line;while ((line = br.readLine()) != null) {sb.append(line).append("\n");}String schemaStr = sb.toString();br.close();fr.close();Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "dev3-yangyunhe-group001");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("topic_lcc"));Schema.Parser parser = new Schema.Parser();Schema schema = parser.parse(schemaStr);Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);try {while (true) {ConsumerRecords<String, byte[]> records = consumer.poll(1000);for (ConsumerRecord<String, byte[]> record : records) {GenericRecord genericRecord = recordInjection.invert(record.value()).get();System.out.println("value = [user.id = " + genericRecord.get("id") + ", " +"user.name = " + genericRecord.get("name") + ", " +"user.age = " + genericRecord.get("age") + "], " +"partition = " + record.partition() + ", " +"offset = " + record.offset());}}} finally {consumer.close();}}}

先运行 KafkaConsumer，没有输出
当运行 KakfaProducer 后，KakfaConsumer 控制台输出：

value = [user.id = 0, user.name = name0, user.age = 22], partition = 2, offset = 662
value = [user.id = 1, user.name = name1, user.age = 22], partition = 1, offset = 663
value = [user.id = 2, user.name = name2, user.age = 22], partition = 0, offset = 663
value = [user.id = 3, user.name = name3, user.age = 22], partition = 2, offset = 663
value = [user.id = 4, user.name = name4, user.age = 22], partition = 1, offset = 664

【Kafka】Kafka 使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化相关推荐

SpringBoot 自定义Kafka消息序列化和反序列化
1. 概述 Kafka传输自定义的DTO对象时,不能像平时一样使用StringSerializer和StringDeserializer.这种情况需要自己实现对应DTO的序列化器和反序列化器.假设现在 ...
Kafka消息序列化和反序列化（下）
欢迎支持笔者新作:<深入理解Kafka:核心设计与实践原理>和<RabbitMQ实战指南>,同时欢迎关注笔者的微信公众号:朱小厮的博客. 欢迎跳转到本文的原文链接:https: ...
Kafka消息序列化和反序列化（上）
欢迎支持笔者新作:<深入理解Kafka:核心设计与实践原理>和<RabbitMQ实战指南>,同时欢迎关注笔者的微信公众号:朱小厮的博客. 欢迎跳转到本文的原文链接:https: ...
kafka python框架_Python中如何使用Apache Avro——Apache的数据序列化系统
了解如何创建和使用基于Apache Avro的数据,以实现更好,更有效的传输. 在这篇文章中,我将讨论Apache Avro,这是一种开源数据序列化系统,Spark,Kafka等工具正在使用该工具进行 ...
[Kafka] Kafka基本架构
[Kafka] Kafka基本架构 [Kafka] Kafka基本架构 [Kafka] Kafka基本架构生产者Producer :生产信息: 消费者Consumer :订阅主题.消费信息: 代理B ...
[Big Data - Kafka] kafka学习笔记：知识点整理
一.为什么需要消息系统 1.解耦: 允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束. 2.冗余:消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险.许 ...
linux上卸载kafka,kafka安装在linux上的安装
kafka安装第一关 java的安装捞得嘛,不谈第二关 zookeeper的安装及配置 1. 直接打开Apach zookeeper进行下载 Tips: source 是源文件,需要编译后才能继 ...
部署kafka kafka的service容器和zookeeper kafka客户端 Elasticsearch的客户端
创建network docker network create -d overlay --attachable loc_net docker stack up -c kafka.yml kafk ...
[kafka]kafka集群实践
环境 ip hostname server_id 192.168.1.111 UAT04 2 192.168.1.112 UAT03 1 192.168.1.102 UAT05 3 配置hosts: ...

【Kafka】Kafka 使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化

1.概述

2.依赖

3.测试类

【Kafka】Kafka 使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化相关推荐

最新文章

热门文章