1，什么是序列化和反序列化
當(dāng)需要將數(shù)據(jù)存入文件或者通過網(wǎng)絡(luò)發(fā)送出去時(shí)，需將數(shù)據(jù)對(duì)象轉(zhuǎn)化為字節(jié)流，即對(duì)數(shù)據(jù)序列化，而將字節(jié)流還原成對(duì)象即反序列化。常用的序列化框架有Apache Avro、Twitter的Thrift和Google的Protobuf。它們之間的顯著區(qū)別就是Avro支持IDL（接口描述語(yǔ)言）和Json描述schema，而Thrift和Protobuf只支持IDL描述Schema。本文將討論Kafka中如何利用Apache avro框架序列/反序列化。
2，Apache avro序列化步驟
第一步：準(zhǔn)備jar包
從官網(wǎng)下載avro-1.7.7.jar?和?avro-tools-1.7.7.jar兩個(gè)jar包，放到指定文件目錄。下載地址為http://www.trieuvan.com/apache/avro/avro-1.7.7/java/。
第二步：編寫Schema描述文件
可以用Json和IDL描述schema，本文以Json描述schema文件。如創(chuàng)建stock.avsc的文本文件，文件內(nèi)容如下：
{
? ? "namespace": "example.avro",
? ? "type": "record",
? ? "name": "Stock",
? ? "fields": [
? ? ? ? {"name": "stockCode", "type": "string"},
? ? ? ? {"name": "stockName",? "type": "string"},
? ? ? ? {"name": "tradeTime", "type": "long"},
? ? ? ? {"name": "preClosePrice", "type": "float"},
? ? ? ? {"name": "openPrice", "type": "float"},
? ? ? ? {"name": "currentPrice", "type": "float"},
? ? ? ? {"name": "highPrice", "type": "float"},
? ? ? ? {"name": "lowPrice", "type": "float"}
? ? ]
}
第三步：自動(dòng)生成Java 對(duì)象類
執(zhí)行如下命令：
java -jar avro-tools-1.7.7.jar compile schema stock.avsc java .
文中的 . 表示當(dāng)前目錄
第四步：在maven工程中引入下面jar包
<dependency>
? <groupId>org.apache.avro</groupId>
? <artifactId>avro</artifactId>
? <version>1.7.7</version>
</dependency>
第五步：將自動(dòng)生成的java 對(duì)象源碼放入到工程中，并注意修改代碼中的包路徑等?？梢岳貌檎襡xample.avro 替換成自己工程的包路徑；
第六步：自定義序列化和反序列化類
序列化類：
package com.bigdata.kafkasender;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Map;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;
public class AvroSerializer implements Serializer<Stock> {
? ? @Override
? ? public void close() {
? ? }
? ? @Override
? ? public void configure(Map<String, ?> arg0, boolean arg1) {
? ? }
? ? @Override
? ? public byte[] serialize(String topic, Stock data) {
? ? ? ? if (data == null) {
? ? ? ? ? ? return null;
? ? ? ? }
? ? ? ? DatumWriter<Stock> writer = new SpecificDatumWriter<>(data.getSchema());
? ? ? ? ByteArrayOutputStream out = new ByteArrayOutputStream();
? ? ? ? BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(out, null);
? ? ? ? try {
? ? ? ? ? ? writer.write(data, encoder);
? ? ? ? } catch (IOException e) {
? ? ? ? ? ? throw new SerializationException(e.getMessage());
? ? ? ? }
? ? ? ? return out.toByteArray();
? ? }
}
反序列化類：
package com.bigdata.kafkasender;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.Map;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.kafka.common.serialization.Deserializer;
public class AvroDeserializer implements Deserializer<Stock> {
? ? @Override
? ? public void close() {}
? ? @Override
? ? public void configure(Map<String, ?> arg0, boolean arg1) {}
? ? @Override
? ? public Stock deserialize(String topic, byte[] data) {
? ? ? ? if(data == null) {
? ? ? ? ? ? return null;
? ? ? ? }
? ? ? ? Stock stock = new Stock();
? ? ? ? ByteArrayInputStream in = new ByteArrayInputStream(data);
? ? ? ? DatumReader<Stock> userDatumReader = new SpecificDatumReader<>(stock.getSchema());
? ? ? ? BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(in, null);
? ? ? ? try {
? ? ? ? ? ? stock = userDatumReader.read(null, decoder);
? ? ? ? } catch (IOException e) {
? ? ? ? ? ? e.printStackTrace();
? ? ? ? }
? ? ? ? return stock;
? ? }
}
第七步：編寫kafka生產(chǎn)者
package com.bigdata.kafkasender;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
public class KafkaSenderApplication {
? ? public static void main(String[] args) throws ExecutionException, InterruptedException {
? ? ? ? Stock[] stocks = new Stock[100];
? ? ? ? for (int i = 0; i < 100; i++) {
? ? ? ? ? ? stocks[i] = new Stock();
? ? ? ? ? ? stocks[i].setStockCode(String.valueOf(i));
? ? ? ? ? ? stocks[i].setStockName("stock" + i);
? ? ? ? ? ? stocks[i].setTradeTime(System.currentTimeMillis());
? ? ? ? ? ? stocks[i].setPreClosePrice(100.0F);
? ? ? ? ? ? stocks[i].setOpenPrice(88.8F);
? ? ? ? ? ? stocks[i].setCurrentPrice(120.5F);
? ? ? ? ? ? stocks[i].setHighPrice(300.0F);
? ? ? ? ? ? stocks[i].setLowPrice(12.4F);
? ? ? ? }
? ? ? ? Properties props = new Properties();
? ? ? ? props.put("bootstrap.servers", "xx.x.x.xx:9094");
? ? ? ? props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
? ? ? ? props.put("value.serializer", "com.bigdata.kafkasender.AvroSerializer");
? ? ? ? KafkaProducer<String, Stock> producer = new KafkaProducer<>(props);
? ? ? ? Future<RecordMetadata> result = null;
? ? ? ? for (Stock stock : stocks) {
? ? ? ? ? ? ProducerRecord<String, Stock> record = new ProducerRecord<>("avro1", stock);
? ? ? ? ? ? RecordMetadata metadata = producer.send(record).get();
? ? ? ? ? ? StringBuilder sb = new StringBuilder();
? ? ? ? ? ? sb.append("stock: ").append(stock.toString()).append(" has been sent successfully!").append("\n")
? ? ? ? ? ? ? ? ? ? .append("send to partition ").append(metadata.partition())
? ? ? ? ? ? ? ? ? ? .append(", offset = ").append(metadata.offset());
? ? ? ? ? ? System.out.println(sb.toString());
? ? ? ? ? ? Thread.sleep(100);
? ? ? ? }
? ? ? ? producer.close();
? ? }
}
第八步：編寫消費(fèi)者
public static void consumeMessage() {
? ? ? ? Properties props = new Properties();
? ? ? ? /* 定義kakfa 服務(wù)的地址，不需要將所有broker指定上 */
? ? ? ? props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "sxxxx:9094");
? ? ? ? /* 制定consumer group */
? ? ? ? // props.put("group.id", "flume");
? ? ? ? props.put("group.id", "avro1");
? ? ? ? props.put("auto.offset.reset", "earliest");
? ? ? ? /* key的序列化類 */
? ? ? ? props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
? ? ? ? /* value的序列化類 */
? ? ? ? props.put("value.deserializer", "com.bigdata.kafkasender.AvroDeserializer");
? ? ? ? /* 定義consumer */
? ? ? ? KafkaConsumer<String, Stock> consumer = new KafkaConsumer<>(props);
? ? ? ? /* 消費(fèi)者訂閱的topic, 可同時(shí)訂閱多個(gè) */
? ? ? ? consumer.subscribe(Arrays.asList("avro1"));
? ? ? ? System.out.println("begin consume!");
? ? ? ? /* 讀取數(shù)據(jù)，讀取超時(shí)時(shí)間為100ms */
? ? ? ? Duration duration = Duration.ofMillis(100);
? ? ? ? while (true) {
? ? ? ? ? ? ConsumerRecords<String, Stock> records = consumer.poll(duration);
? ? ? ? ? ? for (ConsumerRecord<String, Stock> record : records)
? ? ? ? ? ? ? ? System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(),
? ? ? ? ? ? ? ? ? ? ? ? record.value().toString() + "\n");
? ? ? ? }
? ? }
? ? public static void main(String[] args) {
// produceMessage();
? ? ? ? consumeMessage();
? ? }
