关于 avro 的 maven 工程的搭建以及 avro 的入门知识，可以参考： Apache Avro 入门
1. 定义 schema 文件，并编译 maven 工程生成实体类
schema 文件名称为：stock.avsc，内容如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| 1
2{
3 "namespace": "com.bonc.rdpe.kafka110.beans",
4 "type": "record",
5 "name": "Stock",
6 "fields": [
7 {"name": "stockCode", "type": "string"},
8 {"name": "stockName", "type": "string"},
9 {"name": "tradeTime", "type": "long"},
10 {"name": "preClosePrice", "type": "float"},
11 {"name": "openPrice", "type": "float"},
12 {"name": "currentPrice", "type": "float"},
13 {"name": "highPrice", "type": "float"},
14 {"name": "lowPrice", "type": "float"}
15 ]
16}
17 |
编译 maven 工程生成实体类：
2. 自定义序列化类和反序列化类
(1) 序列化类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
| 1
2package com.bonc.rdpe.kafka110.serializer;
3
4import java.io.ByteArrayOutputStream;
5import java.io.IOException;
6import java.util.Map;
7
8import org.apache.avro.io.BinaryEncoder;
9import org.apache.avro.io.DatumWriter;
10import org.apache.avro.io.EncoderFactory;
11import org.apache.avro.specific.SpecificDatumWriter;
12import org.apache.kafka.common.errors.SerializationException;
13import org.apache.kafka.common.serialization.Serializer;
14
15import com.bonc.rdpe.kafka110.beans.Stock;
16
17/**
18 * @Title AvroSerializer.java
19 * @Description 使用传统的 Avro API 自定义序列化类
20 * @Author YangYunhe
21 * @Date 2018-06-21 16:40:35
22 */
23public class AvroSerializer implements Serializer<Stock> {
24
25 @Override
26 public void close() {}
27
28 @Override
29 public void configure(Map<String, ?> arg0, boolean arg1) {}
30
31 @Override
32 public byte[] serialize(String topic, Stock data) {
33 if(data == null) {
34 return null;
35 }
36 DatumWriter<Stock> writer = new SpecificDatumWriter<>(data.getSchema());
37 ByteArrayOutputStream out = new ByteArrayOutputStream();
38 BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(out, null);
39 try {
40 writer.write(data, encoder);
41 }catch (IOException e) {
42 throw new SerializationException(e.getMessage());
43 }
44 return out.toByteArray();
45 }
46
47}
48 |
(2) 反序列化类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
| 1
2package com.bonc.rdpe.kafka110.deserializer;
3
4import java.io.ByteArrayInputStream;
5import java.io.IOException;
6import java.util.Map;
7
8import org.apache.avro.io.BinaryDecoder;
9import org.apache.avro.io.DatumReader;
10import org.apache.avro.io.DecoderFactory;
11import org.apache.avro.specific.SpecificDatumReader;
12import org.apache.kafka.common.serialization.Deserializer;
13
14import com.bonc.rdpe.kafka110.beans.Stock;
15
16/**
17 * @Title AvroDeserializer.java
18 * @Description 使用传统的 Avro API 自定义反序列类
19 * @Author YangYunhe
20 * @Date 2018-06-21 17:19:40
21 */
22public class AvroDeserializer implements Deserializer<Stock> {
23
24 @Override
25 public void close() {}
26
27 @Override
28 public void configure(Map<String, ?> arg0, boolean arg1) {}
29
30 @Override
31 public Stock deserialize(String topic, byte[] data) {
32 if(data == null) {
33 return null;
34 }
35 Stock stock = new Stock();
36 ByteArrayInputStream in = new ByteArrayInputStream(data);
37 DatumReader<Stock> userDatumReader = new SpecificDatumReader<>(stock.getSchema());
38 BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(in, null);
39 try {
40 stock = userDatumReader.read(null, decoder);
41 } catch (IOException e) {
42 e.printStackTrace();
43 }
44 return stock;
45 }
46}
47 |
3. KafkaProducer使用自定义的序列化类发送消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
| 1
2package com.bonc.rdpe.kafka110.producer;
3
4import java.util.Properties;
5
6import org.apache.kafka.clients.producer.KafkaProducer;
7import org.apache.kafka.clients.producer.Producer;
8import org.apache.kafka.clients.producer.ProducerRecord;
9import org.apache.kafka.clients.producer.RecordMetadata;
10
11import com.bonc.rdpe.kafka110.beans.Stock;
12
13/**
14 * @Title TraditionalAvroProducer.java
15 * @Description Kafka Producer 发送avro序列化后的Stock对象
16 * @Author YangYunhe
17 * @Date 2018-06-21 17:41:59
18 */
19public class TraditionalAvroProducer {
20
21 public static void main(String[] args) throws Exception {
22
23 Stock[] stocks = new Stock[100];
24 for(int i = 0; i < 100; i++) {
25 stocks[i] = new Stock();
26 stocks[i].setStockCode(String.valueOf(i));
27 stocks[i].setStockName("stock" + i);
28 stocks[i].setTradeTime(System.currentTimeMillis());
29 stocks[i].setPreClosePrice(100.0F);
30 stocks[i].setOpenPrice(88.8F);
31 stocks[i].setCurrentPrice(120.5F);
32 stocks[i].setHighPrice(300.0F);
33 stocks[i].setLowPrice(12.4F);
34 }
35
36 Properties props = new Properties();
37 props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
38 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
39 // 设置序列化类为自定义的 avro 序列化类
40 props.put("value.serializer", "com.bonc.rdpe.kafka110.serializer.AvroSerializer");
41
42 Producer<String, Stock> producer = new KafkaProducer<>(props);
43
44 for(Stock stock : stocks) {
45 ProducerRecord<String, Stock> record = new ProducerRecord<>("dev3-yangyunhe-topic001", stock);
46 RecordMetadata metadata = producer.send(record).get();
47 StringBuilder sb = new StringBuilder();
48 sb.append("stock: ").append(stock.toString()).append(" has been sent successfully!").append("\n")
49 .append("send to partition ").append(metadata.partition())
50 .append(", offset = ").append(metadata.offset());
51 System.out.println(sb.toString());
52 Thread.sleep(100);
53 }
54
55 producer.close();
56 }
57}
58 |
4. KafkaConsumer使用自定义的反序列化类接收消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
| 1
2package com.bonc.rdpe.kafka110.consumer;
3
4import java.util.Collections;
5import java.util.Properties;
6
7import org.apache.kafka.clients.consumer.ConsumerRecord;
8import org.apache.kafka.clients.consumer.ConsumerRecords;
9import org.apache.kafka.clients.consumer.KafkaConsumer;
10
11import com.bonc.rdpe.kafka110.beans.Stock;
12
13/**
14 * @Title TraditionalAvroConsumer.java
15 * @Description Kafka Consumer 解析avro序列化后的Stock对象
16 * @Author YangYunhe
17 * @Date 2018-06-21 17:43:03
18 */
19public class TraditionalAvroConsumer {
20
21 public static void main(String[] args) {
22
23 Properties props = new Properties();
24 props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
25 props.put("group.id", "dev3-yangyunhe-group001");
26 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
27 // 设置反序列化类为自定义的avro反序列化类
28 props.put("value.deserializer","com.bonc.rdpe.kafka110.deserializer.AvroDeserializer");
29 KafkaConsumer<String, Stock> consumer = new KafkaConsumer<>(props);
30
31 consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"));
32
33 try {
34 while(true) {
35 ConsumerRecords<String, Stock> records = consumer.poll(100);
36 for(ConsumerRecord<String, Stock> record : records) {
37 Stock stock = record.value();
38 System.out.println(stock.toString());
39 }
40 }
41 }finally {
42 consumer.close();
43 }
44 }
45}
46 |
5. 测试结果
运行生产者代码后控制台输出：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| 1
2stock: {"stockCode": "0", "stockName": "stock0", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4} has been sent successfully!
3send to partition 0, offset = 552
4stock: {"stockCode": "1", "stockName": "stock1", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4} has been sent successfully!
5send to partition 2, offset = 551
6stock: {"stockCode": "2", "stockName": "stock2", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4} has been sent successfully!
7send to partition 1, offset = 551
8stock: {"stockCode": "3", "stockName": "stock3", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4} has been sent successfully!
9send to partition 0, offset = 553
10stock: {"stockCode": "4", "stockName": "stock4", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4} has been sent successfully!
11send to partition 2, offset = 552
12
13......
14 |
运行消费者代码后控制台输出：
1
2
3
4
5
6
7
8
9
| 1
2{"stockCode": "0", "stockName": "stock0", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4}
3{"stockCode": "1", "stockName": "stock1", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4}
4{"stockCode": "2", "stockName": "stock2", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4}
5{"stockCode": "3", "stockName": "stock3", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4}
6{"stockCode": "4", "stockName": "stock4", "tradeTime": 1529631848353, "preClosePrice": 100.0, "openPrice": 88.8, "currentPrice": 120.5, "highPrice": 300.0, "lowPrice": 12.4}
7
8......
9 |
