普通java项目集成kafka方式

风过叶飘林

目录

• 依赖
• Kafka配置读取类
• producer
• consumer
• 使用
• 总结

现在假设一种需求，我方业务系统要与某服务平台通过kafka交互，异步获取服务，而系统架构可能老旧，不是spring cloud桶，不是spring boot，只是java普通项目或者 java web项目

依赖

1. <dependency>
2. <groupId>org.apache.kafka</groupId>
3. <artifactId>kafka-clients</artifactId>
4. <version>3.1.0</version>
5. </dependency>
7. <dependency>
8. <groupId>org.apache.kafka</groupId>
9. <artifactId>kafka_2.11</artifactId>
10. <version>2.4.1</version>
11. </dependency>

复制代码

Kafka配置读取类

本文后边没用到，直接填配置了，简单点

但如果生产需要，还是有这个类比较好，可以从不同配置中读取，同时给个默认值

1. import org.apache.commons.lang3.StringUtils;
2. import org.slf4j.Logger;
3. import org.slf4j.LoggerFactory;
4. import java.io.File;
5. import java.io.FileInputStream;
6. import java.io.IOException;
7. import java.util.Properties;
9. /**
10. * kafka配置读取类
11. *
12. * @author zzy
13. */
14. public class KafkaProperties {
16. private static final Logger LOG = LoggerFactory.getLogger(KafkaProperties.class);
18. private static Properties serverProps = new Properties();
20. private static Properties clientProps = new Properties();
22. private static Properties producerProps = new Properties();
24. private static Properties consumerProps = new Properties();
26. private static KafkaProperties instance = null;
28. private KafkaProperties() {
30. String filePath = System.getProperty("user.dir") + File.separator
31. + "kafkaConf" + File.separator;
33. File file;
34. FileInputStream fis = null;
35. try {
36. file = new File(filePath + "producer.properties");
37. if (file.exists()) {
38. fis = new FileInputStream(filePath + "producer.properties");
39. producerProps.load(fis);
40. }
42. file = new File(filePath + "consumer.properties");
43. if (file.exists()) {
44. fis = new FileInputStream(filePath + "consumer.properties");
45. consumerProps.load(fis);
46. }
48. file = new File(filePath + "server.properties");
49. if (file.exists()) {
50. fis = new FileInputStream(filePath + "server.properties");
51. serverProps.load(fis);
52. }
54. file = new File(filePath + "client.properties");
55. if (file.exists()) {
56. fis = new FileInputStream(filePath + "client.properties");
57. clientProps.load(fis);
58. }
60. } catch (Exception e) {
62. LOG.error("init kafka props error." + e.getMessage());
64. } finally {
66. if (fis != null) {
67. try {
68. fis.close();
69. } catch (IOException e) {
70. LOG.error("close kafka properties fis error." + e);
71. }
72. }
74. }
76. }
78. /**
79. * 获取懒汉式单例
80. */
81. public static synchronized KafkaProperties getInstance() {
82. if (instance == null) {
83. instance = new KafkaProperties();
84. }
86. return instance;
87. }
90. /**
91. * 获取配置，获取不到时使用参数的默认配置
92. */
93. public String getValue(String key, String defaultValue) {
94. String value;
96. if (StringUtils.isEmpty(key)) {
97. LOG.error("key is null or empty");
98. }
99. value = getPropsValue(key);
101. if (value == null) {
102. LOG.warn("kafka property getValue return null, the key is " + key);
103. value = defaultValue;
104. }
105. LOG.info("kafka property getValue, key:" + key + ", value:" + value);
107. return value;
108. }
111. private String getPropsValue(String key) {
112. String value = serverProps.getProperty(key);
114. if (value == null) {
115. value = producerProps.getProperty(key);
116. }
118. if (value == null) {
119. value = consumerProps.getProperty(key);
120. }
122. if (value == null) {
123. value = clientProps.getProperty(key);
124. }
126. return value;
127. }
128. }

复制代码

producer

1. import org.apache.kafka.clients.producer.Callback;
2. import org.apache.kafka.clients.producer.KafkaProducer;
3. import org.apache.kafka.clients.producer.ProducerRecord;
4. import org.apache.kafka.clients.producer.RecordMetadata;
5. import org.slf4j.Logger;
6. import org.slf4j.LoggerFactory;
7. import java.util.Properties;
8. import java.util.concurrent.ExecutionException;
10. /**
11. * kafka producer
12. * @author zzy
13. */
14. public class KafkaProFormal {
16. public static final Logger LOG = LoggerFactory.getLogger(KafkaProFormal.class);
18. private Properties properties = new Properties();
20. private final String bootstrapServers = "bootstrap.servers";
21. private final String clientId = "client.id";
22. private final String keySerializer = "key.serializer";
23. private final String valueSerializer = "value.serializer";
24. //private final String securityProtocol = "security.protocol";
25. //private final String saslKerberosServiceName = "sasl.kerberos.service.name";
26. //private final String kerberosDomainName = "kerberos.domain.name";
27. private final String maxRequestSize = "max.request.size";
29. private KafkaProducer<String, String> producer;
31. private volatile static KafkaProFormal kafkaProFormal;
33. private KafkaProFormal(String servers) {
34. properties.put(bootstrapServers, servers);
35. properties.put(keySerializer, "org.apache.kafka.common.serialization.StringSerializer");
36. properties.put(valueSerializer, "org.apache.kafka.common.serialization.StringSerializer");
38. producer = new KafkaProducer<String, String>(properties);
39. }
41. public static KafkaProFormal getInstance(String servers) {
42. if(kafkaProFormal == null) {
43. synchronized(KafkaProFormal.class) {
44. if(kafkaProFormal == null) {
45. kafkaProFormal = new KafkaProFormal(servers);
46. }
47. }
48. }
50. return kafkaProFormal;
51. }
53. public void sendStringWithCallBack(String topic, String message, boolean asyncFlag) {
54. ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, message);
55. long startTime = System.currentTimeMillis();
56. if(asyncFlag) {
57. //异步发送
58. producer.send(record, new KafkaCallBack(startTime, message));
59. } else {
60. //同步发送
61. try {
62. producer.send(record, new KafkaCallBack(startTime, message)).get();
63. } catch (InterruptedException e) {
64. LOG.error("InterruptedException occured : {0}", e);
65. } catch (ExecutionException e) {
66. LOG.error("ExecutionException occured : {0}", e);
67. }
68. }
69. }
70. }
72. class KafkaCallBack implements Callback {
73. private static Logger LOG = LoggerFactory.getLogger(KafkaCallBack.class);
75. private String key;
77. private long startTime;
79. private String message;
81. KafkaCallBack(long startTime, String message) {
82. this.startTime = startTime;
83. this.message = message;
84. }
86. @Override
87. public void onCompletion(RecordMetadata metadata, Exception exception) {
88. long elapsedTime = System.currentTimeMillis() - startTime;
90. if(metadata != null) {
91. LOG.info("Record(" + key + "," + message + ") sent to partition(" + metadata.partition()
92. + "), offset(" + metadata.offset() + ") in " + elapsedTime + " ms.");
93. } else {
94. LOG.error("metadata is null." + "Record(" + key + "," + message + ")", exception);
95. }
96. }
97. }

复制代码

consumer

1. import kafka.utils.ShutdownableThread;
2. import org.apache.kafka.clients.consumer.ConsumerRecords;
3. import org.apache.kafka.clients.consumer.KafkaConsumer;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import java.time.Duration;
7. import java.util.Properties;
8. import java.util.Set;
10. /**
11. * kafka consumer
12. * @author zzy
13. */
14. public abstract class KafkaConFormal extends ShutdownableThread {
16. private static final Logger LOG = LoggerFactory.getLogger(KafkaConFormal.class);
18. private Set<String> topics;
20. private final String bootstrapServers = "bootstrap.servers";
21. private final String groupId = "group.id";
22. private final String keyDeserializer = "key.deserializer";
23. private final String valueDeserializer = "value.deserializer";
24. private final String enableAutoCommit = "enable.auto.commit";
25. private final String autoCommitIntervalMs = "auto.commit.interval.ms";
26. private final String sessionTimeoutMs = "session.timeout.ms";
28. private KafkaConsumer<String, String> consumer;
30. public KafkaConFormal(String topic) {
31. super("KafkaConsumerExample", false);
33. topics.add(topic);
35. Properties props = new Properties();
36. props.put(bootstrapServers, "your servers");
37. props.put(groupId, "TestGroup");
38. props.put(enableAutoCommit, "true");
39. props.put(autoCommitIntervalMs, "1000");
40. props.put(sessionTimeoutMs, "30000");
41. props.put(keyDeserializer, "org.apache.kafka.common.serialization.StringDeserializer");
42. props.put(valueDeserializer, "org.apache.kafka.common.serialization.StringDeserializer");
44. consumer = new KafkaConsumer<>(props);
45. }
47. /**
48. * subscribe and handle the msg
49. */
50. @Override
51. public void doWork() {
52. consumer.subscribe(topics);
53. ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
55. dealRecords(records);
56. }
58. /**
59. * 实例化consumer时，进行对消费信息的处理
60. * @param records records
61. */
62. public abstract void dealRecords(ConsumerRecords<String, String> records);
64. public void setTopics(Set<String> topics) {
65. this.topics = topics;
66. }
67. }

复制代码

使用

1. KafkaProFormal producer = KafkaProFormal.getInstance("kafka server1.1.1.1:9092,2.2.2.2:9092");
3. KafkaConFormal consumer = new KafkaConFormal("consume_topic") {
4. @Override
5. public void dealRecords(ConsumerRecords<String, String> records) {
6. for (ConsumerRecord<String, String> record: records) {
7. producer.sendStringWithCallBack("target_topic", record.value(), true);
8. }
9. }
10. };
12. consumer.start();

复制代码

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持晓枫资讯。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

隆静曼

路过，支持一下

翁良

感谢楼主，顶。

伤心欲绝2017

感谢楼主分享。

123476162

顶顶更健康！！！