Kafka 中使用 Avro 序列化框架(二):使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化

使用传统的 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篇

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏小灰灰

Spring之RestTemplate使用小结一

作为一个Java后端,需要通过HTTP请求其他的网络资源可以说是一个比较常见的case了;一般怎么做呢?

1.2K3
来自专栏Java帮帮-微信公众号-技术文章全总结

JSP与EL表达式重点学习笔记(1)

Jsp&el表达式 JSP指令 ? JSP指令概述 JSP指令的格式:<%@指令名 attr1=”” attr2=”” %>,一般都会把JSP指令放到JSP文件...

3639
来自专栏微信公众号:Java团长

自己手写一个SpringMVC框架

前端框架很多,但没有一个框架称霸,后端框架现在Spring已经完成大一统。所以学习Spring是Java程序员的必修课。

882
来自专栏DT乱“码”

读取.properties配置文件工具类

package com.test.utils; import java.io.FileOutputStream; import java.io.IOExce...

2278
来自专栏人工智能LeadAI

使用强大的 Mockito 来测试你的代码

这篇教程介绍了如何使用 Mockito 框架来给软件写测试用例。 1、预备知识 如果需要往下学习,你需要先理解 Junit 框架中的单元测试。 如果你不熟悉 J...

1.5K6
来自专栏精讲JAVA

接口方法上的注解无法被 @Aspect 声明的切面拦截的原因分析

在Spring中使用MyBatis的Mapper接口自动生成时,用一个自定义的注解标记在Mapper接口的方法中,再利用@Aspect定义一个切面,拦截这个注解...

3234
来自专栏技术记录

shiro权限控制(二):分布式架构中shiro的实现

前言: 前段时间在搭建公司游戏框架安全验证的时候,就想到之前web最火的shiro框架,虽然后面实践发现在netty中不太适用,最后自己模仿shiro写了一个缩...

6007
来自专栏精讲JAVA

接口方法上的注解无法被 @Aspect 声明的切面拦截的原因分析

在Spring中使用MyBatis的Mapper接口自动生成时,用一个自定义的注解标记在Mapper接口的方法中,再利用@Aspect定义一个切面,拦截这个注解...

2412
来自专栏SpringBoot 核心技术

第三十七章:基于SpringBoot架构以及参数装载完成接口安全认证

45110
来自专栏cmazxiaoma的架构师之路

SpringBoot之路(二)之Web进阶

3254

扫码关注云+社区