文章目录
背景
实际开发中,不同的topic可能来自不同的集群,所以就需要配置不同的kafka数据源,基于springboot自动配置的思想,最终通过配置文件的配置,自动生成生产者及消费者的配置。
核心配置
自动化配置类
importcom.example.kafka.autoconfig.CustomKafkaDataSourceRegister;importcom.example.kafka.autoconfig.kafkaConsumerConfig;importorg.springframework.beans.BeansException;importorg.springframework.beans.factory.BeanFactory;importorg.springframework.beans.factory.BeanFactoryAware;importorg.springframework.beans.factory.config.SmartInstantiationAwareBeanPostProcessor;importorg.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;importorg.springframework.boot.context.properties.EnableConfigurationProperties;importorg.springframework.context.annotation.Configuration;importorg.springframework.context.annotation.Import;importorg.springframework.kafka.annotation.EnableKafka;@EnableKafka@Configuration(
proxyBeanMethods =false)@ConditionalOnWebApplication@EnableConfigurationProperties({kafkaConsumerConfig.class})@Import({CustomKafkaDataSourceRegister.class})publicclassMyKafkaAutoConfigurationimplementsBeanFactoryAware,SmartInstantiationAwareBeanPostProcessor{publicMyKafkaAutoConfiguration(){}publicvoidsetBeanFactory(BeanFactory beanFactory)throwsBeansException{
beanFactory.getBean(CustomKafkaDataSourceRegister.class);}}
注册生产者、消费者核心bean到spring
public void afterPropertiesSet() {
Map<String, ConsumerConfigWrapper> factories = kafkaConsumerConfig.getFactories();
if (factories != null && !factories.isEmpty()) {
factories.forEach((factoryName, consumerConfig) -> {
KafkaProperties.Listener listener = consumerConfig.getListener();
Integer concurrency = consumerConfig.getConcurrency();
// 创建监听容器工厂
ConcurrentKafkaListenerContainerFactory<String, String> containerFactory = createKafkaListenerContainerFactory(consumerConfig.buildProperties(), listener, concurrency);
// 注册到容器
if (!beanFactory.containsBean(factoryName)) {
beanFactory.registerSingleton(factoryName, containerFactory);
}
});
}
Map<String, KafkaProperties.Producer> templates = kafkaProducerConfig.getTemplates();
if (!ObjectUtils.isEmpty(templates)) {
templates.forEach((templateName, producerConfig) -> {
//registerBean(beanFactory, templateName, KafkaTemplate.class, propertyValues);
//注册spring bean的两种方式
registerBeanWithConstructor(beanFactory, templateName, KafkaTemplate.class, producerFactoryValues(producerConfig.buildProperties()));
});
}
}
配置spring.factories
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.example.kafka.MyKafkaAutoConfiguration
yml配置
spring:
kafka:
multiple:
consumer:
factories:
test-factory:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
bootstrap-servers: 192.168.56.112:9092
group-id: group_a
concurrency: 25
fetch-min-size: 1048576
fetch-max-wait: 3000
listener:
type: batch
properties:
spring-json-trusted-packages: '*'
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
auto-offset-reset: latest
producer:
templates:
test-template:
bootstrap-servers: 192.168.56.112:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
使用
源码仓库
本文转载自: https://blog.csdn.net/qq_37362891/article/details/139391610
版权归原作者 it噩梦 所有, 如有侵权,请联系我们删除。
版权归原作者 it噩梦 所有, 如有侵权,请联系我们删除。