springboot如何配置多kafka
作者:是菜菜的小严惜哎
这篇文章主要介绍了springboot如何配置多kafka问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
springboot配置多kafka
kafka,说起这个玩意,大家应该都不陌生,不知道啥是kafka的直接去搜索
就像MySQL的使用一样,我们在用kafka的时候,也会碰到需要使用多个kafka的情况,比如我从kafkaA的一个topic里消费出数据,进行一次处理,然后我要写入到kafkaB的topic里
从网上找了很多配置多kafka的教程,但是都不大好使,后来还是找到了,加上我自己改了点点东西,贴出来和大家分享一下~
首先是pom文件,kafka的依赖
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
配置文件
spring: kafka: listener: concurrency: 10 one: # kafka地址 bootstrap-servers: 192.168.217.117:9092 producer: # 生产者每次发送消息的时间间隔(毫秒) linger-ms: 5000 # 单条消息最大值(字节) max-request-size: 16384 # 每次批量发送消息的数量 batch-size: 16384 # 缓存区大小 buffer-memory: 33554432 # 队列 topic: test consumer: # 是否自动提交 enable-auto-commit: true # 队列 topic: topic # group ID group-id: consumer two: # kafka地址 bootstrap-servers: 192.168.217.160:9092 producer: # 生产者每次发送消息的时间间隔(毫秒) linger-ms: 100 # 单条消息最大值(单位为字节,且大小指的是经过序列化后的大小) max-request-size: 16384 # 每次批量发送消息的数量 batch-size: 16384 # 缓存区大小 buffer-memory: 33554432 # 队列 topic: test consumer: # 是否自动提交 enable-auto-commit: true # 队列 topic: test # group ID group-id: consumer
有了配置文件当然就要读取配置文件了
先读取第一个kafka
@EnableKafka @Configuration public class KafkaConfigOne { @Value("${spring.kafka.one.bootstrap-servers}") private String bootstrapServers; @Value("${spring.kafka.one.consumer.enable-auto-commit}") private boolean enableAutoCommit; @Value("${spring.kafka.one.consumer.group-id}") private String groupId; @Value("${spring.kafka.one.producer.linger-ms}") private Integer lingerMs; @Value("${spring.kafka.one.producer.max-request-size}") private Integer maxRequestSize; @Value("${spring.kafka.one.producer.batch-size}") private Integer batchSize; @Value("${spring.kafka.one.producer.buffer-memory}") private Integer bufferMemory; @Bean public KafkaTemplate<String, String> kafkaOneTemplate() { return new KafkaTemplate<>(producerFactory()); } @Bean KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaOneContainerFactory() { ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setConcurrency(3); factory.getContainerProperties().setPollTimeout(3000); return factory; } private ProducerFactory<String, String> producerFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); } public ConsumerFactory<Integer, String> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } private Map<String, Object> producerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.LINGER_MS_CONFIG,lingerMs); props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize); props.put(ProducerConfig.BATCH_SIZE_CONFIG,batchSize); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,bufferMemory); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return props; } private Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); props.put(ConsumerConfig.GROUP_ID_CONFIG,groupId); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return props; } }
接着读取第二个kafka
@EnableKafka @Configuration public class KafkaConfigTwo { @Value("${spring.kafka.two.bootstrap-servers}") private String bootstrapServers; @Value("${spring.kafka.two.consumer.enable-auto-commit}") private boolean enableAutoCommit; @Value("${spring.kafka.two.consumer.group-id}") private String groupId; @Value("${spring.kafka.two.producer.linger-ms}") private Integer lingerMs; @Value("${spring.kafka.two.producer.max-request-size}") private Integer maxRequestSize; @Value("${spring.kafka.two.producer.batch-size}") private Integer batchSize; @Value("${spring.kafka.two.producer.buffer-memory}") private Integer bufferMemory; @Bean public KafkaTemplate<String, String> kafkaTwoTemplate() { return new KafkaTemplate<>(producerFactory()); } @Bean KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaTwoContainerFactory() { ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setConcurrency(3); factory.getContainerProperties().setPollTimeout(3000); return factory; } private ProducerFactory<String, String> producerFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); } public ConsumerFactory<Integer, String> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } private Map<String, Object> producerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.LINGER_MS_CONFIG,lingerMs); props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize); props.put(ProducerConfig.BATCH_SIZE_CONFIG,batchSize); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,bufferMemory); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return props; } private Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); props.put(ConsumerConfig.GROUP_ID_CONFIG,groupId); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return props; } }
PS:路径什么的大家可以改哈
使用:
@Autowired @Qualifier("kafkaTwoTemplate") private KafkaTemplate kafkaTwoTemplate; kafkaTwoTemplate.send("topic", "message");
直接注入使用就可以,但是,在这儿注意,因为配置了多个kafka,所以需要进行区分,此处我使用@Autowired和@Qualifier连用,大家也可以使用@DataSource,这个就是多数据源的注解而已,无所谓,按照个人习惯进行使用就OK
当然,还有消费者
@KafkaListener(topics = {"#{'${spring.kafka.two.consumer.topic}'}"}, containerFactory = "kafkaTwoContainerFactory") public void listenerTwo (String data) { System.out.println(data); }
总结
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。