使用传统的 avro API 自定义序列化类和反序列化类比较麻烦，需要根据 schema 生成实体类，需要调用 avro 的 API 实现 对象到 byte[] 和 byte[] 到对象的转化，而那些方法看上去比较繁琐，幸运的是，Twitter 开源的类库 Bijection 对传统的 Avro API 进行了封装了和优化，让我们可以方便的实现以上操作。
1. 添加 Bijection 类库的依赖，并新建一个 schema 文件
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"}
]
}
2. KafkaProducer 使用 Bijection 类库发送序列化后的消息
package com.bonc.rdpe.kafka110.producer;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
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.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;
/**
* @Title BijectionProducer.java
* @Description KafkaProducer 使用 Bijection 类库发送序列化后的消息
* @Author YangYunhe
* @Date 2018-06-22 10:42:06
*/
public class BijectionProducer {
public static void main(String[] args) 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", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
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<>("dev3-yangyunhe-topic001", avroRecordBytes);
producer.send(record);
Thread.sleep(1000);
}
producer.close();
}
}
3. KafkaConsumer 使用 Bijection 类库来反序列化消息
package com.bonc.rdpe.kafka110.consumer;
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.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import com.bonc.rdpe.kafka110.producer.BijectionProducer;
import com.twitter.bijection.Injection;
import com.twitter.bijection.avro.GenericAvroCodecs;
/**
* @Title BijectionConsumer.java
* @Description KafkaConsumer 使用 Bijection 类库来反序列化消息
* @Author YangYunhe
* @Date 2018-06-22 11:10:29
*/
public class BijectionConsumer {
public static void main(String[] args) 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", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
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("dev3-yangyunhe-topic001"));
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();
}
}
}
4. 测试结果
先运行 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中使用Avro编码消息：Producter篇
在Kafka中使用Avro编码消息：Consumer篇
