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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
package cn.ac.iie.dao;
import cn.ac.iie.bean.voipSipFromToLog.RouteRelationLog;
import cn.ac.iie.bean.voipSipOrigin.SipOriginALL;
import cn.ac.iie.common.RealtimeCountConfig;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.log4j.Logger;
import java.text.SimpleDateFormat;
import java.util.LinkedList;
import java.util.Properties;
/**
* 发送数据的kafka工具类
*/
public class KafkaDB {
private static Logger logger = Logger.getLogger(KafkaDB.class);
private static Producer<String, String> producer;
private static KafkaDB kafkaDB;
private KafkaDB() {
getProducer();
}
public static KafkaDB getInstance() {
if (kafkaDB == null) {
kafkaDB = new KafkaDB();
}
return kafkaDB;
}
public void siporiLog2KafkaFromSipInsertBoltDC(LinkedList<String> sipOriJsonS) {
long time = System.currentTimeMillis() / 1000L;
for (String sipOriJson : sipOriJsonS) {
try {
SipOriginALL sipOriginLog = JSONObject.parseObject(sipOriJson, SipOriginALL.class);
sipOriginLog.setStat_time(time);
producer.send(new ProducerRecord<>(RealtimeCountConfig.KAFKA_SIP_COMPLEMENT_TOPIC, JSONObject.toJSONString(sipOriginLog)));
} catch (Exception e) {
e.printStackTrace();
}
}
producer.flush();
}
public void routeRelatLog2KafkaFromSipInsertBoltDC(LinkedList<String> routeJsonS) {
long time = System.currentTimeMillis() / 1000L;
for (String routeJson : routeJsonS) {
try {
RouteRelationLog routeRelationLog = JSONObject.parseObject(routeJson, RouteRelationLog.class);
routeRelationLog.setTimestamp(time);
producer.send(new ProducerRecord<>(RealtimeCountConfig.KAFKA_ROUTE_RELATION_TOPIC, JSONObject.toJSONString(routeRelationLog)));
} catch (Exception e) {
e.printStackTrace();
}
}
producer.flush();
}
/**
* 根据kafka生产者配置信息初始化kafka消息生产者,只初始化一次
*/
private void getProducer() {
Properties properties = new Properties();
properties.put("bootstrap.servers", RealtimeCountConfig.BOOTSTRAP_OUTPUT_SERVERS);
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("acks", "1");
properties.put("linger.ms", "2");
properties.put("request.timeout.ms", 20000);
properties.put("batch.size", 262144);
properties.put("buffer.memory", 33554432);
producer = new KafkaProducer<String, String>(properties);
}
}
|
