聊聊Spring?Boot如何配置多個(gè)Kafka數(shù)據(jù)源
更新時(shí)間：2023年10月20日 09:35:50 作者：Mr?Tang
這篇文章主要介紹了Spring?Boot配置多個(gè)Kafka數(shù)據(jù)源的相關(guān)知識(shí),包括生產(chǎn)者、消費(fèi)者配置,本文結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧
一、配置文件
application.properties配置文件如下
#kafka多數(shù)據(jù)源配置 #kafka數(shù)據(jù)源一，日志審計(jì)推送 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數(shù)據(jù)源二，動(dòng)環(huán)數(shù)據(jù)消費(fèi) 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>
三、生產(chǎn)者、消費(fèi)者配置
1.第一個(gè)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.第二個(gè)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;
}
}四.生產(chǎn)者
@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";
}
}五.消費(fèi)者
@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());
}
}備注：
生產(chǎn)者消費(fèi)者代碼參考鏈接，開發(fā)同學(xué)需要以實(shí)際情況按要求自己變更下代碼即可：
Spring Boot 集成多個(gè) Kafka_springboot集成多個(gè)kafka
到此這篇關(guān)于Spring Boot配置多個(gè)Kafka數(shù)據(jù)源的文章就介紹到這了,更多相關(guān)Spring Boot配置Kafka數(shù)據(jù)源內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Java獲取當(dāng)前時(shí)間String類型和Date類型方式
這篇文章主要介紹了Java獲取當(dāng)前時(shí)間String類型和Date類型方式,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-07-07
Java?Web項(xiàng)目中解決中文亂碼方法總結(jié)(三種最新方法)
這篇文章主要介紹了Java?Web項(xiàng)目中解決中文亂碼方法總結(jié),本文給大家分享三種最新解決方法，需要的朋友可以參考下2022-06-06
這篇文章主要介紹了java如何獲取實(shí)體類的屬性名和屬性值問題,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-11-11
SpringBoot中JPA實(shí)現(xiàn)Sort排序的三種方式小結(jié)
這篇文章主要介紹了SpringBoot中JPA實(shí)現(xiàn)Sort排序的三種方式小結(jié)，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-11-11
Spring Boot 全局異常處理策略設(shè)計(jì)之@ExceptionHandler 與&nb
SpringBoot全局異常處理策略設(shè)計(jì)@ExceptionHandler與@ControllerAdvice生效原理源碼解析,本文通過實(shí)例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧2026-01-01
Java實(shí)現(xiàn)中文字符串與unicode互轉(zhuǎn)工具類
這篇文章主要為大家詳細(xì)介紹了Java實(shí)現(xiàn)中文字符串與unicode互轉(zhuǎn)的工具類，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-04-04
搭建Spring cloud 非阻塞式微服務(wù)架構(gòu)的實(shí)現(xiàn)
本文介紹搭建非阻塞式微服務(wù)架構(gòu),采用Spring Gateway、Spring Security、JWT與Redis,集成EurekaServer、Auth-service、User-service等服務(wù),實(shí)現(xiàn)用戶認(rèn)證、日志跟蹤及統(tǒng)一入口管理,感興趣的可以了解一下2025-08-08
這篇文章主要介紹了Jenkins自動(dòng)部署Net Core過程圖解,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-12-12
這篇文章主要介紹了SpringBoot登錄攔截配置詳解(實(shí)測可用)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-07-07
通過反射實(shí)現(xiàn)Java下的委托機(jī)制代碼詳解
這篇文章主要介紹了通過反射實(shí)現(xiàn)Java下的委托機(jī)制代碼詳解，具有一定借鑒價(jià)值，需要的朋友可以參考下。2017-12-12