聊聊Spring?Boot如何配置多個Kafka數據源
更新時間：2023年10月20日 09:35:50 作者：Mr?Tang
這篇文章主要介紹了Spring?Boot配置多個Kafka數據源的相關知識,包括生產者、消費者配置,本文結合實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧
一、配置文件
application.properties配置文件如下
#kafka多數據源配置 #kafka數據源一，日志審計推送 spring.kafka.one.bootstrap-servers=172.19.12.109:32182 spring.kafka.one.producer.retries=0 spring.kafka.one.producer.properties.max.block.ms=5000 #kafka數據源二，動環(huán)數據消費 spring.kafka.two.bootstrap-servers=172.19.12.109:32182 spring.kafka.two.producer.retries=0 spring.kafka.two.producer.properties.max.block.ms=5000 spring.kafka.two.consumer.group-id=bw-convert-data spring.kafka.two.consumer.enable-auto-commit=true
二、pom依賴
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
三、生產者、消費者配置
1.第一個kakfa
package com.gstanzer.convert.config;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.*;
import java.util.HashMap;
import java.util.Map;
@EnableKafka
@Configuration
public class KafkaOneConfig {
@Value("${spring.kafka.one.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.one.producer.retries}")
private String retries;
@Value("${spring.kafka.one.producer.properties.max.block.ms}")
private String maxBlockMs;
@Bean
public KafkaTemplate<String, String> kafkaOneTemplate() {
return new KafkaTemplate<>(producerFactory());
}
private ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
private Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
}2.第二個kakfa
package com.gstanzer.convert.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import java.util.HashMap;
import java.util.Map;
@Configuration
@EnableKafka
public class KafkaTwoConfig {
@Value("${spring.kafka.two.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.two.producer.retries}")
private String retries;
@Value("${spring.kafka.two.producer.properties.max.block.ms}")
private String maxBlockMs;
@Value("${spring.kafka.two.consumer.group-id}")
private String groupId;
@Value("${spring.kafka.two.consumer.enable-auto-commit}")
private boolean enableAutoCommit;
@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.RETRIES_CONFIG, retries);
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
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.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
}四.生產者
@Controller
public class TestController {
@Autowired
private KafkaTemplate kafkaOneTemplate;
@Autowired
private KafkaTemplate kafkaTwoTemplate;
@RequestMapping("/send")
@ResponseBody
public String send() {
final String TOPIC = "TOPIC_1";
kafkaOneTemplate.send(TOPIC, "kafka one");
kafkaTwoTemplate.send(TOPIC, "kafka two");
return "success";
}
}五.消費者
@Component
public class KafkaConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumer.class);
final String TOPIC = "TOPIC_1";
// containerFactory 的值要與配置中 KafkaListenerContainerFactory 的 Bean 名相同
@KafkaListener(topics = {TOPIC}, containerFactory = "kafkaOneContainerFactory")
public void listenerOne(ConsumerRecord<?, ?> record) {
LOGGER.info(" kafka one 接收到消息：{}", record.value());
}
@KafkaListener(topics = {TOPIC}, containerFactory = "kafkaTwoContainerFactory")
public void listenerTwo(ConsumerRecord<?, ?> record) {
LOGGER.info(" kafka two 接收到消息：{}", record.value());
}
}備注：
生產者消費者代碼參考鏈接，開發(fā)同學需要以實際情況按要求自己變更下代碼即可：
Spring Boot 集成多個 Kafka_springboot集成多個kafka
到此這篇關于Spring Boot配置多個Kafka數據源的文章就介紹到這了,更多相關Spring Boot配置Kafka數據源內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了Java獲取當前時間String類型和Date類型方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-07-07
這篇文章主要介紹了Java?Web項目中解決中文亂碼方法總結,本文給大家分享三種最新解決方法，需要的朋友可以參考下2022-06-06
這篇文章主要介紹了java如何獲取實體類的屬性名和屬性值問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
這篇文章主要介紹了SpringBoot中JPA實現Sort排序的三種方式小結，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-11-11
Spring Boot 全局異常處理策略設計之@ExceptionHandler 與&nb
SpringBoot全局異常處理策略設計@ExceptionHandler與@ControllerAdvice生效原理源碼解析,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧2026-01-01
這篇文章主要為大家詳細介紹了Java實現中文字符串與unicode互轉的工具類，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-04-04
本文介紹搭建非阻塞式微服務架構,采用Spring Gateway、Spring Security、JWT與Redis,集成EurekaServer、Auth-service、User-service等服務,實現用戶認證、日志跟蹤及統(tǒng)一入口管理,感興趣的可以了解一下2025-08-08
這篇文章主要介紹了Jenkins自動部署Net Core過程圖解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-12-12
這篇文章主要介紹了SpringBoot登錄攔截配置詳解(實測可用)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-07-07
這篇文章主要介紹了通過反射實現Java下的委托機制代碼詳解，具有一定借鑒價值，需要的朋友可以參考下。2017-12-12