Gitlink
/
gitlink-notification-system
0
0
Fork 4
Code Issues Pull Requests Releases 1 Wiki Activity

修改kafa监听初始化配置

This commit is contained in:
曾伟 2021-09-18 15:11:26 +08:00
parent ddb7098080
commit cfb4fe6280
4 changed files with 31 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java → common/src/main/java/cn/org/gitlink/notification/common/config/KafkaConsumerConfig.java
View File

@ -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);
}
}

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

@ -34,18 +34,15 @@ public class KafkaUtil {
@Value("${spring.kafka.producer.bootstrap_servers:#{null}}")
private String kafkaServer;
@Value("${spring.kafka.producer.topic:#{null}")
private String topic;
@Value("${spring.kafka.producer.mail_topic:#{null}")
private String mailTopic;
@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
@ -70,20 +67,14 @@ public class KafkaUtil {
adminClient = KafkaAdminClient.create(props);
// 初始化topics
// TODO: 如果后面新增topic，需要在这里添加
if (null !=partitions && null != replicationFactor) {
List<NewTopic> topics = new LinkedList<>();
if (null != topic) {
topics.add(new NewTopic(topic, partitions, replicationFactor));
}
if (null != mailTopic) {
topics.add(new NewTopic(mailTopic, partitions, replicationFactor));
}
List<String> topics = Arrays.asList(topicString.split(","));
if (!topics.isEmpty()) {
this.createTopic(topics);
List<NewTopic> newTopics = new LinkedList<>();
topics.forEach(topic -> {
newTopics.add(new NewTopic(topic, partitions, replicationFactor));
});
this.createTopic(newTopics);
}
}

2
executor/src/main/resources/application.yml.example
View File

@ -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

2
writer/src/main/resources/application.yml.example
View File

@ -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