Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitc0a03d2

Browse files
committed
Producer: add json custom serializer
1 parent3af386f commitc0a03d2

File tree

3 files changed

+68
-2
lines changed

3 files changed

+68
-2
lines changed

‎Producer/src/main/java/com/telkomdev/producer/App.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
packagecom.telkomdev.producer;
1818

1919
importcom.telkomdev.producer.model.Product;
20+
importcom.telkomdev.producer.serializer.ProductJsonSerializer;
2021
importcom.telkomdev.producer.serializer.ProductProtobufSerializer;
2122
importorg.apache.kafka.clients.producer.KafkaProducer;
2223
importorg.apache.kafka.clients.producer.Producer;
@@ -55,8 +56,15 @@ public static void main(String[] args) {
5556
// kafka brokers
5657
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokers);
5758
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.ByteArraySerializer.class.getName());
59+
60+
// send String data
5861
//producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
59-
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,ProductProtobufSerializer.class.getName());
62+
63+
// send Protocol Buffer data
64+
//producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ProductProtobufSerializer.class.getName());
65+
66+
// send JSON data
67+
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,ProductJsonSerializer.class.getName());
6068

6169
Producerproducer =newKafkaProducer<String,String>(producerConfig);
6270

@@ -87,7 +95,7 @@ public static void main(String[] args) {
8795
producer.send(record);
8896
input =in.nextLine();
8997
}catch (Exceptionex) {
90-
System.out.println("error send data to kafka: "+ex.getMessage());
98+
System.out.println("error send data to kafka: " +ex.getMessage());
9199
break;
92100
}
93101
}

‎Producer/src/main/java/com/telkomdev/producer/model/Product.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
packagecom.telkomdev.producer.model;
1818

19+
importcom.google.gson.Gson;
20+
importcom.google.gson.GsonBuilder;
1921
importcom.google.protobuf.InvalidProtocolBufferException;
2022
importcom.telkomdev.producer.protojava.ProductProto;
2123

@@ -31,6 +33,8 @@ public class Product {
3133
privateIntegerquantity;
3234
privateList<String>images;
3335

36+
privatefinalGsongson =newGsonBuilder().setPrettyPrinting().create();
37+
3438
publicProduct() {
3539

3640
}
@@ -84,6 +88,16 @@ public String toString() {
8488
'}';
8589
}
8690

91+
publicbyte[]toJson() {
92+
returngson.toJson(this).getBytes();
93+
}
94+
95+
publicstaticProductfromJson(byte[]in) {
96+
Gsong =newGson();
97+
StringjsonString =newString(in);
98+
returng.fromJson(jsonString,Product.class);
99+
}
100+
87101
publicProductProto.ProducttoProto() {
88102
ProductProto.ProductproductOut =ProductProto.Product.newBuilder()
89103
.setID(this.id)
Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
2+
/*
3+
* Copyright 2019 wuriyanto.com
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
packagecom.telkomdev.producer.serializer;
19+
20+
importcom.telkomdev.producer.model.Product;
21+
importorg.apache.kafka.common.serialization.Serializer;
22+
23+
importjava.util.Map;
24+
25+
publicclassProductJsonSerializerimplementsSerializer<Product> {
26+
27+
publicProductJsonSerializer() {
28+
}
29+
30+
@Override
31+
publicvoidconfigure(Map<String, ?>configs,booleanisKey) {
32+
33+
}
34+
35+
@Override
36+
publicbyte[]serialize(Strings,Productproduct) {
37+
returnproduct ==null ?null :product.toJson();
38+
}
39+
40+
@Override
41+
publicvoidclose() {
42+
43+
}
44+
}

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp