聊聊Spring Boot 如何集成多個(gè) Kafka
更新時(shí)間：2023年10月20日 09:33:47 作者：//承續(xù)緣_紀(jì)錄片
這篇文章主要介紹了Spring Boot 集成多個(gè) Kafka的相關(guān)資料,包括配置文件,生成者和消費(fèi)者配置過(guò)程,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì),需要的朋友參考下吧
一、配置文件
application.yml
spring:
kafka:
one:
bootstrap-servers: IP:PORT
consumer:
group-id: YOUR_GROUP_ID
enable-auto-commit: true
two:
bootstrap-servers: IP:PORT
consumer:
group-id: YOUR_GROUP_ID
enable-auto-commit: true二、生產(chǎn)者、消費(fèi)者配置
2.1 第一個(gè) Kafka
@EnableKafka
@Configuration
public class KafkaOneConfig {
@Value("${spring.kafka.one.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.one.consumer.group-id}")
private String groupId;
@Value("${spring.kafka.one.consumer.enable-auto-commit}")
private boolean enableAutoCommit;
@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.RETRIES_CONFIG, 0);
props.put(ProducerConfig.ACKS_CONFIG, "1"); // 不能寫成 1
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;
}
}2.2 第二個(gè) Kafka
@Configuration
public class KafkaTwoConfig {
@Value("${spring.kafka.two.bootstrap-servers}")
private String bootstrapServers;
@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, 0);
props.put(ProducerConfig.ACKS_CONFIG, "1");
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());
}
}運(yùn)行結(jié)果
c.k.s.consumer.KafkaConsumer : kafka one 接收到消息：kafka one
c.k.s.consumer.KafkaConsumer : kafka two 接收到消息：kafka two
到此這篇關(guān)于聊聊Spring Boot 如何集成多個(gè) Kafka的文章就介紹到這了,更多相關(guān)Spring Boot集成多個(gè) Kafka內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
相關(guān)文章
SpringBoot文件訪問(wèn)映射如何實(shí)現(xiàn)
這篇文章主要介紹了SpringBoot文件訪問(wèn)映射如何實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-02-02
這篇文章主要為大家介紹了java?class?name實(shí)例深入精講,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪
2023-09-09 
最近在本地開發(fā)測(cè)試的時(shí)候，遇到一個(gè)表單重復(fù)提交的現(xiàn)象。本文主要介紹了springmvc 防止表單重復(fù)提交的兩種方法，感興趣的可以了解一下
2021-08-08 
Log4j有三個(gè)主要的組件/對(duì)象：Loggers(記錄器)，Appenders (輸出源)和Layouts(布局)。這里可簡(jiǎn)單理解為日志類別，日志要輸出的地方和日志以何種形式輸出，今天通過(guò)本文給大家分享Java日志框架log4j的相關(guān)知識(shí)，感興趣的朋友一起看看吧
2021-08-08 
這篇文章主要介紹了SpringAOP中的通知Advice詳解,Spring 的 AOP 功能中一個(gè)關(guān)鍵概念是通知Advice與切點(diǎn)Pointcut表達(dá)式相關(guān)聯(lián)在特定節(jié)點(diǎn)織入一些邏輯，Spring 提供了五種類型的通知,需要的朋友可以參考下
2023-08-08 
詳解SpringBoot中的tomcat優(yōu)化和修改
這篇文章主要介紹了詳解SpringBoot中的tomcat優(yōu)化和修改，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
2020-09-09 
簡(jiǎn)介Java的Hibernate框架中的Session和持久化類
這篇文章主要介紹了Java的Hibernate框架中的Session和持久化類,Hibernate是Java的SSH三大web開發(fā)框架之一,需要的朋友可以參考下
2015-12-12 
Idea?springboot?springCloud熱加載熱調(diào)試兩種常用方式
這篇文章主要介紹了Idea?springboot?springCloud熱加載熱調(diào)試常用的兩種方式,在項(xiàng)目開發(fā)的過(guò)程中，需要修改調(diào)試的時(shí)候偶每次都需要重啟項(xiàng)目浪費(fèi)時(shí)間，下面是我整理的兩種常用的兩種方式，需要的朋友可以參考下
2023-04-04 
Spring框架自身并沒(méi)有實(shí)現(xiàn)緩存解決方案，但是從3.1開始定義了org.springframework.cache.Cache和org.springframework.cache.CacheManager接口，提供對(duì)緩存功能的聲明，能夠與多種流行的緩存實(shí)現(xiàn)集成。這篇文章主要介紹了Spring 緩存抽象 ,需要的朋友可以參考下
2018-09-09