diff --git a/executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java b/common/src/main/java/cn/org/gitlink/notification/common/config/KafkaConsumerConfig.java similarity index 57% rename from executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java rename to common/src/main/java/cn/org/gitlink/notification/common/config/KafkaConsumerConfig.java index 6ab75aa..dda9057 100644 --- a/executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java +++ b/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 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 consumerConfigs() { Map 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); } } diff --git a/common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java b/common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java index 78c737b..d315f31 100644 --- a/common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java +++ b/common/src/main/java/cn/org/gitlink/notification/common/config/KafkaProducerConfig.java @@ -33,12 +33,17 @@ public class KafkaProducerConfig { @Bean ProducerFactory producerConfigs() { Map 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); } } diff --git a/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java b/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java index d682afa..efcf318 100644 --- a/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java +++ b/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 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 topics = Arrays.asList(topicString.split(",")); + if (!topics.isEmpty()) { + List newTopics = new LinkedList<>(); + topics.forEach(topic -> { + newTopics.add(new NewTopic(topic, partitions, replicationFactor)); + }); + this.createTopic(newTopics); + } + } + } public void createTopic(Collection newTopics) { diff --git a/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java b/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java index d2b4a09..da6b029 100644 --- a/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java +++ b/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/EmailJobsListener.java @@ -30,12 +30,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 @Qualifier(value = "GitlinkKafkaUtil") private KafkaUtil kafkaUtil; @@ -47,7 +41,6 @@ public class EmailJobsListener { Boolean flag = emailJobsService.createEmailJob(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) { diff --git a/executor/src/main/resources/application.yml.example b/executor/src/main/resources/application.yml.example index 3f89c5c..b82de5c 100644 --- a/executor/src/main/resources/application.yml.example +++ b/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 diff --git a/reader/src/main/resources/application.yml.example b/reader/src/main/resources/application.yml.example index ed88882..89cfd92 100644 --- a/reader/src/main/resources/application.yml.example +++ b/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 diff --git a/writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java b/writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java index e6719e6..f66d490 100644 --- a/writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java +++ b/writer/src/main/java/cn/org/gitlink/notification/writer/controller/EmailJobsController.java @@ -9,7 +9,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; @@ -21,7 +20,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 @@ -33,12 +31,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 @Qualifier(value = "GitlinkKafkaUtil") private KafkaUtil kafkaUtil; @@ -71,7 +63,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) { diff --git a/writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java b/writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java index 8056ccb..6b0b818 100644 --- a/writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java +++ b/writer/src/main/java/cn/org/gitlink/notification/writer/controller/NotificationController.java @@ -12,7 +12,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; @@ -24,7 +23,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 @@ -34,12 +32,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 @Qualifier(value = "GitlinkKafkaUtil") 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) { diff --git a/writer/src/main/resources/application.yml.example b/writer/src/main/resources/application.yml.example index 88fc52e..124c9ea 100644 --- a/writer/src/main/resources/application.yml.example +++ b/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