修改kafa监听初始化配置

1.初始化时创建topic
2.初始化时没有kafka相关配置信息则不进行kafka连接

common/src/main/java/cn/org/gitlink/notification/common/config/KafkaConsumerConfig.java

@@ -1,4 +1,4 @@
-package cn.org.gitlink.notification.executor.core.config;
+package cn.org.gitlink.notification.common.config;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -20,17 +20,21 @@ public class KafkaConsumerConfig {
     @Value("${spring.kafka.consumer.bootstrap_servers:#{null}}")
     private String servers;
 
-    @Value("${spring.kafka.consumer.auto_offset_reset}")
+    @Value("${spring.kafka.consumer.auto_offset_reset:#{null}}")
     private String autoOffsetReset;
 
-    @Value("${spring.kafka.consumer.max_poll_records}")
+    @Value("${spring.kafka.consumer.max_poll_records:#{null}")
     private String maxPollRecords;
 
-    @Value("${spring.kafka.consumer.topic}")
+    @Value("${spring.kafka.consumer.topic:#{null}}")
     private String topic;
 
     @Bean
     public ConcurrentKafkaListenerContainerFactory<String, String> consumerListenerFactory() {
+
+        // 如果当前配置信息内没有kafka consumer则不创建监听工厂
+        if (null == servers) return null;
+
         ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
         factory.setConsumerFactory(consumerConfigs());
         factory.setRecordFilterStrategy(record -> record.topic().toLowerCase().equals(this.topic));
@@ -40,12 +44,16 @@ public class KafkaConsumerConfig {
     @Bean
     public ConsumerFactory<String, Object> consumerConfigs() {
         Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
-        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
-        props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+
+        // 如果当前配置信息内没有kafka consumer，此处解决没有设置配置信息而引起的空指针异常
+        if (null != servers) {
+            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
+            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
+            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
+            props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
+            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        }
         return new DefaultKafkaConsumerFactory<>(props);
     }
 }

common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java

@@ -33,12 +33,17 @@ public class KafkaProducerConfig {
     @Bean
     ProducerFactory<String, String> producerConfigs() {
         Map<String, Object> props = new HashMap<>();
-        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        props.put(ProducerConfig.RETRIES_CONFIG, retries);
-        props.put(ProducerConfig.ACKS_CONFIG, "all");
-        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
-        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+        // 如果当前配置信息内没有kafka producer相关配置则不做参数设置，此处解决没有设置配置信息而引起的空指针异常
+        if (null != bootstrapServers) {
+            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+            props.put(ProducerConfig.RETRIES_CONFIG, retries);
+            props.put(ProducerConfig.ACKS_CONFIG, "all");
+            props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
+            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        }
+
         return new DefaultKafkaProducerFactory<>(props);
     }
 }

common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java

@@ -14,10 +14,7 @@ import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import javax.annotation.PostConstruct;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
@@ -37,6 +34,15 @@ public class KafkaUtil {
     @Value("${spring.kafka.producer.bootstrap_servers:#{null}}")
     private String kafkaServer;
 
+    @Value("${spring.kafka.producer.partitions:#{null}}")
+    private Integer partitions;
+
+    @Value("${spring.kafka.producer.replication_factor:#{null}}")
+    private Short replicationFactor;
+
+    @Value("${spring.kafka.producer.topics:''}")
+    private String topicString;
+
     private AdminClient adminClient;
 
     @Autowired
@@ -46,16 +52,32 @@ public class KafkaUtil {
      * 初始化AdminClient
      * '@PostConstruct该注解被用来修饰一个非静态的void（）方法。
      * 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。
-     * PostConstruct在构造函数之后执行，init（）方法之前执行。
+     * PostConstruct在构造函数之后执行，init（）方法之前执行。ls
      */
     @PostConstruct
     private void initAdminClient() {
         Map<String, Object> props = new HashMap<>(1);
+
+        // 如果当前配置信息内没有kafka producer相关配置则不对adminClient做初始化
         if (kafkaServer == null) {
             return;
         }
+
         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
         adminClient = KafkaAdminClient.create(props);
+
+        // 初始化topics
+        if (null !=partitions && null != replicationFactor) {
+            List<String> topics = Arrays.asList(topicString.split(","));
+            if (!topics.isEmpty()) {
+                List<NewTopic> newTopics = new LinkedList<>();
+                topics.forEach(topic -> {
+                    newTopics.add(new NewTopic(topic, partitions, replicationFactor));
+                });
+                this.createTopic(newTopics);
+            }
+        }
+
     }
 
     public void createTopic(Collection<NewTopic> newTopics) {

executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java

@@ -29,12 +29,6 @@ public class EmailJobsListener {
     @Value("${spring.kafka.producer.topic_new_email_remind}")
     private String gitlinkNewEmailRemindTopic;
 
-    @Value("${spring.kafka.producer.partitions}")
-    private Integer partitions;
-
-    @Value("${spring.kafka.producer.replication_factor}")
-    private Short replicationFactor;
-
     @Autowired
     private KafkaUtil kafkaUtil;
 
@@ -45,7 +39,6 @@ public class EmailJobsListener {
             Boolean flag = emailJobsService.sendEmail(newEmailJobVo);
             //if the message is inserted successfully, send a new email-job message to kafka
             if (flag){
-                kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNewEmailRemindTopic, partitions, replicationFactor)));
                 kafkaUtil.sendMessage(gitlinkNewEmailRemindTopic, JSONObject.toJSONString(newEmailJobVo));
             }
         } catch (Exception e) {

executor/src/main/resources/application.yml.example

@@ -13,6 +13,8 @@ spring:
       replication_factor: 1
       partitions: 3
       topic_new_email_remind: topic-gitlink-new-email-remind
+      # 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
+      topics: ${spring.kafka.producer.topic_new_email_remind}
     consumer:
       bootstrap_servers: kafka1:9092,kafka2:9092
       group_id: group-gitlink-notification

reader/src/main/resources/application.yml.example

@@ -18,12 +18,6 @@ spring:
       min-idle: 0
     timeout: 1000
 
-  kafka:
-    producer:
-      bootstrap_servers: kafka1:9092,kafka2:9092
-      retries: 5
-      batch_size: 16384
-
   datasource:
     driver-class-name: com.mysql.jdbc.Driver
     url: jdbc:mysql://mysql:3306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true&useSSL=false

writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java

@@ -10,7 +10,6 @@ import cn.org.gitlink.notification.model.dao.entity.vo.NewEmailJobVo;
 import com.alibaba.fastjson.JSONObject;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.beans.BeanUtils;
@@ -22,7 +21,6 @@ import org.springframework.validation.BindingResult;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
-import java.util.Arrays;
 import java.util.Map;
 
 @RestController
@@ -35,12 +33,6 @@ public class EmailJobsController {
     @Value("${spring.kafka.producer.topic_email}")
     private String gitlinkEmailTopic;
 
-    @Value("${spring.kafka.producer.partitions}")
-    private Integer partitions;
-
-    @Value("${spring.kafka.producer.replication_factor}")
-    private Short replicationFactor;
-
     @Autowired
     private KafkaUtil kafkaUtil;
 
@@ -72,7 +64,6 @@ public class EmailJobsController {
         newEmailJobVo.setPlatform(platform);
 
         try {
-            kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkEmailTopic, partitions, replicationFactor)));
             kafkaUtil.sendMessage(gitlinkEmailTopic, JSONObject.toJSONString(newEmailJobVo));
             return DataPacketUtil.jsonSuccessResult();
         } catch (Exception e) {

writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java

@@ -13,7 +13,6 @@ import cn.org.gitlink.notification.model.service.notification.SysNotificationSer
 import com.alibaba.fastjson.JSONObject;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.beans.BeanUtils;
@@ -25,7 +24,6 @@ import org.springframework.validation.BindingResult;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
-import java.util.Arrays;
 import java.util.Map;
 
 @RestController
@@ -36,12 +34,6 @@ public class NotificationController {
     @Value("${spring.kafka.producer.topic}")
     private String gitlinkNotificationTopic;
 
-    @Value("${spring.kafka.producer.partitions}")
-    private Integer partitions;
-
-    @Value("${spring.kafka.producer.replication_factor}")
-    private Short replicationFactor;
-
     @Autowired
     private KafkaUtil kafkaUtil;
 
@@ -79,7 +71,6 @@ public class NotificationController {
         newSysNotificationVo.setPlatform(platform);
 
         try {
-            kafkaUtil.createTopic(Arrays.asList(new NewTopic(gitlinkNotificationTopic, partitions, replicationFactor)));
             kafkaUtil.sendMessage(gitlinkNotificationTopic, JSONObject.toJSONString(newSysNotificationVo));
             return DataPacketUtil.jsonSuccessResult();
         } catch (Exception e) {

writer/src/main/resources/application.yml.example

@@ -14,6 +14,8 @@ spring:
       partitions: 3
       topic: topic-gitlink-notification
       topic_email: topic-gitlink-email
+      # 需要初始化的topic, 添加新的topic后要在这里加上, 用逗号隔开
+      topics: ${spring.kafka.producer.topic}, ${spring.kafka.producer.topic_email}
 
   redis:
     database: 0