springboot使用@KafkaListener監(jiān)聽(tīng)多個(gè)kafka配置實(shí)現(xiàn)
更新時(shí)間：2024年04月09日 09:36:45 作者：道不平
當(dāng)服務(wù)中需要監(jiān)聽(tīng)多個(gè)kafka時(shí),?需要配置多個(gè)kafka,本文主要介紹了springboot使用@KafkaListener監(jiān)聽(tīng)多個(gè)kafka配置實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下
背景
使用springboot整合kafka時(shí), springboot默認(rèn)讀取配置文件中 spring.kafka...配置初始化kafka, 使用@KafkaListener時(shí)指定topic即可, 當(dāng)服務(wù)中需要監(jiān)聽(tīng)多個(gè)kafka時(shí), 需要配置多個(gè)kafka, 這種方式不適用
方案
可以手動(dòng)讀取不同kafka配置信息, 創(chuàng)建不同的Kafka 監(jiān)聽(tīng)容器工廠, 使用@KafkaListener時(shí)指定相應(yīng)的容器工廠, 代碼如下:
1. 導(dǎo)入依賴
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
2. yml配置
kafka:
# 默認(rèn)消費(fèi)者配置
default-consumer:
# 自動(dòng)提交已消費(fèi)offset
enable-auto-commit: true
# 自動(dòng)提交間隔時(shí)間
auto-commit-interval: 1000
# 消費(fèi)的超時(shí)時(shí)間
poll-timeout: 1500
# 如果Kafka中沒(méi)有初始偏移量，或者服務(wù)器上不再存在當(dāng)前偏移量（例如，因?yàn)樵摂?shù)據(jù)已被刪除）自動(dòng)將該偏移量重置成最新偏移量
auto.offset.reset: latest
# 消費(fèi)會(huì)話超時(shí)時(shí)間(超過(guò)這個(gè)時(shí)間consumer沒(méi)有發(fā)送心跳,就會(huì)觸發(fā)rebalance操作)
session.timeout.ms: 120000
# 消費(fèi)請(qǐng)求超時(shí)時(shí)間
request.timeout.ms: 180000
# 1號(hào)kafka配置
test1:
bootstrap-servers: xxxx:xxxx,xxxx:xxxx,xxxx:xxxx
consumer:
group-id: xxx
sasl.mechanism: xxxx
security.protocol: xxxx
sasl.jaas.config: xxxx
# 2號(hào)kafka配置
test2:
bootstrap-servers: xxxx:xxxx,xxxx:xxxx,xxxx:xxxx
consumer:
group-id: xxx
sasl.mechanism: xxxx
security.protocol: xxxx
sasl.jaas.config: xxxx
3. 容器工廠配置
package com.zhdx.modules.backstage.config;
import com.google.common.collect.Maps;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
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.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import java.util.Map;
/**
* kafka監(jiān)聽(tīng)容器工廠配置
* <p>
* 拓展其他消費(fèi)者配置只需配置指定的屬性和bean即可
*/
@EnableKafka
@Configuration
@RefreshScope
public class KafkaListenerContainerFactoryConfig {
/**
* test1 kafka配置
*/
@Value("${kafka.test1.bootstrap-servers}")
private String test1KafkaServerUrls;
@Value("${kafka.test1.consumer.group-id}")
private String test1GroupId;
@Value("${kafka.test1.consumer.sasl.mechanism}")
private String test1SaslMechanism;
@Value("${kafka.test1.consumer.security.protocol}")
private String test1SecurityProtocol;
@Value("${kafka.test1.consumer.sasl.jaas.config}")
private String test1SaslJaasConfig;
/**
* test2 kafka配置
*/
@Value("${kafka.test2.bootstrap-servers}")
private String test2KafkaServerUrls;
@Value("${kafka.test2.consumer.group-id}")
private String test2GroupId;
@Value("${kafka.test2.consumer.sasl.mechanism}")
private String test2SaslMechanism;
@Value("${kafka.test2.consumer.security.protocol}")
private String test2SecurityProtocol;
@Value("${kafka.test2.consumer.sasl.jaas.config}")
private String test2SaslJaasConfig;
/**
* 默認(rèn)消費(fèi)者配置
*/
@Value("${kafka.default-consumer.enable-auto-commit}")
private boolean enableAutoCommit;
@Value("${kafka.default-consumer.poll-timeout}")
private int pollTimeout;
@Value("${kafka.default-consumer.auto.offset.reset}")
private String autoOffsetReset;
@Value("${kafka.default-consumer.session.timeout.ms}")
private int sessionTimeoutMs;
@Value("${kafka.default-consumer.request.timeout.ms}")
private int requestTimeoutMs;
/**
* test1消費(fèi)者配置
*/
public Map<String, Object> test1ConsumerConfigs() {
Map<String, Object> props = getDefaultConsumerConfigs();
// broker server地址
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, test1KafkaServerUrls);
// 消費(fèi)者組
props.put(ConsumerConfig.GROUP_ID_CONFIG, test1GroupId);
// 加密
props.put(SaslConfigs.SASL_MECHANISM, test1SaslMechanism);
props.put("security.protocol", test1SecurityProtocol);
// 賬號(hào)密碼
props.put(SaslConfigs.SASL_JAAS_CONFIG, test1SaslJaasConfig);
return props;
}
/**
* test2消費(fèi)者配置
*/
public Map<String, Object> test2ConsumerConfigs() {
Map<String, Object> props = getDefaultConsumerConfigs();
// broker server地址
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, test2KafkaServerUrls);
// 消費(fèi)者組
props.put(ConsumerConfig.GROUP_ID_CONFIG, test2GroupId);
// 加密
props.put(SaslConfigs.SASL_MECHANISM, test2SaslMechanism);
props.put("security.protocol", test2SecurityProtocol);
// 賬號(hào)密碼
props.put(SaslConfigs.SASL_JAAS_CONFIG, test2SaslJaasConfig);
return props;
}
/**
* 默認(rèn)消費(fèi)者配置
*/
private Map<String, Object> getDefaultConsumerConfigs() {
Map<String, Object> props = Maps.newHashMap();
// 自動(dòng)提交（按周期）已消費(fèi)offset 批量消費(fèi)下設(shè)置false
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
// 消費(fèi)會(huì)話超時(shí)時(shí)間(超過(guò)這個(gè)時(shí)間consumer沒(méi)有發(fā)送心跳,就會(huì)觸發(fā)rebalance操作)
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
// 消費(fèi)請(qǐng)求超時(shí)時(shí)間
props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs);
// 序列化
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
// 如果Kafka中沒(méi)有初始偏移量，或者服務(wù)器上不再存在當(dāng)前偏移量（例如，因?yàn)樵摂?shù)據(jù)已被刪除）自動(dòng)將該偏移量重置成最新偏移量
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
return props;
}
/**
* 消費(fèi)者工廠類(lèi)
*/
public ConsumerFactory<String, String> initConsumerFactory(Map<String, Object> consumerConfigs) {
return new DefaultKafkaConsumerFactory<>(consumerConfigs);
}
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> initKafkaListenerContainerFactory(
Map<String, Object> consumerConfigs) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(initConsumerFactory(consumerConfigs));
// 是否開(kāi)啟批量消費(fèi)
factory.setBatchListener(false);
// 消費(fèi)的超時(shí)時(shí)間
factory.getContainerProperties().setPollTimeout(pollTimeout);
return factory;
}
/**
* 創(chuàng)建test1 Kafka 監(jiān)聽(tīng)容器工廠。
*
* @return KafkaListenerContainerFactory<ConcurrentMessageListenerContainer < String, String>> 返回的 KafkaListenerContainerFactory 對(duì)象
*/
@Bean(name = "test1KafkaListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> test1KafkaListenerContainerFactory() {
Map<String, Object> consumerConfigs = this.test1ConsumerConfigs();
return initKafkaListenerContainerFactory(consumerConfigs);
}
/**
* 創(chuàng)建test2 Kafka 監(jiān)聽(tīng)容器工廠。
*
* @return KafkaListenerContainerFactory<ConcurrentMessageListenerContainer < String, String>> 返回的 KafkaListenerContainerFactory 對(duì)象
*/
@Bean(name = "test2KafkaListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> test2KafkaListenerContainerFactory() {
Map<String, Object> consumerConfigs = this.test2ConsumerConfigs();
return initKafkaListenerContainerFactory(consumerConfigs);
}
}4. @KafkaListener使用
package com.zhdx.modules.backstage.kafka;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* kafka監(jiān)聽(tīng)器
*/
@Slf4j
@Component
public class test1KafkaListener {
@KafkaListener(containerFactory = "test1KafkaListenerContainerFactory", topics = "xxx")
public void handleHyPm(ConsumerRecord<String, String> record) {
log.info("消費(fèi)到topic xxx消息：{}", JSON.toJSONString(record.value()));
}
}到此這篇關(guān)于springboot使用@KafkaListener監(jiān)聽(tīng)多個(gè)kafka配置實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)springboot 監(jiān)聽(tīng)多個(gè)kafka內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
相關(guān)文章
Springboot升級(jí)到2.7.2結(jié)合nacos遇到的坑及解決
這篇文章主要介紹了Springboot升級(jí)到2.7.2結(jié)合nacos遇到的坑及解決,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-06-06
使用JDBC4.0操作XML類(lèi)型的字段(保存獲取xml數(shù)據(jù))的方法
jdbc4.0最重要的特征是支持xml數(shù)據(jù)類(lèi)型，接下來(lái)通過(guò)本文重點(diǎn)給大家介紹如何使用jdbc4.0操作xml類(lèi)型的字段，對(duì)jdbc4.0 xml相關(guān)知識(shí)感興趣的朋友一起看下吧2016-08-08
SpringBoot切面攔截@PathVariable參數(shù)及拋出異常的全局處理方式
這篇文章主要介紹了SpringBoot切面攔截@PathVariable參數(shù)及拋出異常的全局處理方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-08-08
java.lang.Void 與 void的比較及使用方法介紹
這篇文章主要介紹了java.lang.Void 與 void的比較及使用方法介紹，小編覺(jué)得挺不錯(cuò)的，這里給大家分享一下，需要的朋友可以參考。2017-10-10
這篇文章主要為大家詳細(xì)介紹了Java HttpURLConnection使用方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-11-11
Java使用System.currentTimeMillis()方法計(jì)算程序運(yùn)行時(shí)間的示例代碼
System.currentTimeMillis() 方法的返回類(lèi)型為 long ，表示毫秒為單位的當(dāng)前時(shí)間，文中通過(guò)示例代碼介紹了計(jì)算 String 類(lèi)型與 StringBuilder 類(lèi)型拼接字符串的耗時(shí)情況，對(duì)Java計(jì)算程序運(yùn)行時(shí)間相關(guān)知識(shí)感興趣的朋友一起看看吧2022-03-03
這篇文章主要介紹了使用Spring Cache設(shè)置緩存條件操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-09-09
Java并發(fā)Futures和Callables類(lèi)實(shí)例詳解
Callable對(duì)象返回Future對(duì)象,該對(duì)象提供監(jiān)視線程執(zhí)行的任務(wù)進(jìn)度的方法, Future對(duì)象可用于檢查Callable的狀態(tài),然后線程完成后從Callable中檢索結(jié)果,這篇文章給大家介紹Java并發(fā)Futures和Callables類(lèi)的相關(guān)知識(shí),感興趣的朋友一起看看吧2024-05-05
java去除空格、標(biāo)點(diǎn)符號(hào)的方法實(shí)例
這篇文章主要給大家介紹了關(guān)于java去除空格、標(biāo)點(diǎn)符號(hào)的相關(guān)資料，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-09-09
Java中實(shí)現(xiàn)多線程關(guān)鍵詞整理(總結(jié))
這篇文章主要介紹了Java中實(shí)現(xiàn)多線程關(guān)鍵詞整理，非常不錯(cuò)，具有參考借鑒價(jià)值,需要的朋友可以參考下2017-05-05