springboot之配置雙kafka全過程
更新時間：2023年04月14日 15:46:30 作者：竇再興
這篇文章主要介紹了springboot之配置雙kafka全過程，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
springboot配置雙kafka
使用spring boot 2.0.8.RELEASE 版本
引入Maven kafka jar、準(zhǔn)備兩個kafka；
<dependency> ? ? <groupId>org.springframework.kafka</groupId> ? ? <artifactId>spring-kafka</artifactId> </dependency>
配置yml配置文件
spring: ? kafka: ? ? bootstrap-servers: 180.167.180.242:9092 #kafka的訪問地址，多個用","隔開 ? ? consumer: ? ? ? enable-auto-commit: true ? ? ? group-id: kafka #群組ID ? outkafka: ? ? bootstrap-servers: localhost:9092 #kafka的訪問地址，多個用","隔開 ? ? consumer: ? ? ? enable-auto-commit: true ? ? ? group-id: kafka_1 #群組ID
配置KafkaConfig類
import java.util.HashMap;
import java.util.Map;
?
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.context.annotation.Primary;
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.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
?
@Configuration
@EnableKafka
public class KafkaConfig {
? ? @Value("${spring.kafka.bootstrap-servers}")
? ? private String innerServers;
? ? @Value("${spring.kafka.consumer.group-id}")
? ? private String innerGroupid;
? ? @Value("${spring.kafka.consumer.enable-auto-commit}")
? ? private String innerEnableAutoCommit;
?
? ? @Bean
? ? @Primary//理解為默認(rèn)優(yōu)先選擇當(dāng)前容器下的消費者工廠
? ? KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
? ? ? ? ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
? ? ? ? factory.setConsumerFactory(consumerFactory());
? ? ? ? factory.setConcurrency(3);
? ? ? ? factory.getContainerProperties().setPollTimeout(3000);
? ? ? ? return factory;
? ? }
?
? ? @Bean//第一個消費者工廠的bean
? ? public ConsumerFactory<Integer, String> consumerFactory() {
? ? ? ? return new DefaultKafkaConsumerFactory<>(consumerConfigs());
? ? }
?
? ? @Bean
? ? public Map<String, Object> consumerConfigs() {
? ? ? ? Map<String, Object> props = new HashMap<>();
? ? ? ? props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers);
? ? ? ? props.put(ConsumerConfig.GROUP_ID_CONFIG, innerGroupid);
? ? ? ? props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, innerEnableAutoCommit);
// ? ? ? ?props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
// ? ? ? ?props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
? ? ? ? props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? return props;
? ? }
? ??
? ? @Bean //生產(chǎn)者工廠配置
? ? public ProducerFactory<String, String> producerFactory() {
? ? ? ? return new DefaultKafkaProducerFactory<>(senderProps());
? ? }
? ??
? ? @Bean //kafka發(fā)送消息模板
? ? public KafkaTemplate<String, String> kafkaTemplate() {
? ? ? ? return new KafkaTemplate<String, String>(producerFactory());
? ? }
? ??
? ? /**
? ? ?* 生產(chǎn)者配置方法
? ? ?*
? ? ?* 生產(chǎn)者有三個必選屬性
? ? ?* <p>
? ? ?* 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
? ? ?* 生產(chǎn)者會從給定的broker里查找到其他broker的信息。不過建議至少提供兩個broker信息，一旦 其中一個宕機(jī)，生產(chǎn)者仍能能夠連接到集群上。
? ? ?* </p>
? ? ?* <p>
? ? ?* 2.key.serializer broker希望接收到的消息的鍵和值都是字節(jié)數(shù)組。 生產(chǎn)者用對應(yīng)的類把鍵對象序列化成字節(jié)數(shù)組。
? ? ?* </p>
? ? ?* <p>
? ? ?* 3.value.serializer 值得序列化方式
? ? ?* </p>
? ? ?*
? ? ?*
? ? ?* @return
? ? ?*/
? ? private Map<String, Object> senderProps() {
? ? ? ? Map<String, Object> props = new HashMap<>();
? ? ? ? props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers);
? ? ? ? /**
? ? ? ? ?* 當(dāng)從broker接收到的是臨時可恢復(fù)的異常時，生產(chǎn)者會向broker重發(fā)消息，但是不能無限
? ? ? ? ?* 制重發(fā)，如果重發(fā)次數(shù)達(dá)到限制值，生產(chǎn)者將不會重試并返回錯誤。
? ? ? ? ?* 通過retries屬性設(shè)置。默認(rèn)情況下生產(chǎn)者會在重試后等待100ms，可以通過 retries.backoff.ms屬性進(jìn)行修改
? ? ? ? ?*/
? ? ? ? props.put(ProducerConfig.RETRIES_CONFIG, 0);
? ? ? ? /**
? ? ? ? ?* 在考慮完成請求之前，生產(chǎn)者要求leader收到的確認(rèn)數(shù)量。這可以控制發(fā)送記錄的持久性。允許以下設(shè)置：
? ? ? ? ?* <ul>
? ? ? ? ?* <li>
? ? ? ? ?* <code> acks = 0 </ code>如果設(shè)置為零，則生產(chǎn)者將不會等待來自服務(wù)器的任何確認(rèn)。該記錄將立即添加到套接字緩沖區(qū)并視為已發(fā)送。在這種情況下，無法保證服務(wù)器已收到記錄，并且
? ? ? ? ?* <code>retries </ code>配置將不會生效（因為客戶端通常不會知道任何故障）。為每條記錄返回的偏移量始終設(shè)置為-1。
? ? ? ? ?* <li> <code> acks = 1 </code>
? ? ? ? ?* 這意味著leader會將記錄寫入其本地日志，但無需等待所有follower的完全確認(rèn)即可做出回應(yīng)。在這種情況下，
? ? ? ? ?* 如果leader在確認(rèn)記錄后立即失敗但在關(guān)注者復(fù)制之前，則記錄將丟失。
? ? ? ? ?* <li><code> acks = all </code>
? ? ? ? ?* 這意味著leader將等待完整的同步副本集以確認(rèn)記錄。這保證了只要至少一個同步副本仍然存活，記錄就不會丟失。這是最強(qiáng)有力的保證。
? ? ? ? ?* 這相當(dāng)于acks = -1設(shè)置
? ? ? ? ?*/
? ? ? ? props.put(ProducerConfig.ACKS_CONFIG, "1");
? ? ? ? /**
? ? ? ? ?* 當(dāng)有多條消息要被發(fā)送到統(tǒng)一分區(qū)是，生產(chǎn)者會把他們放到統(tǒng)一批里。kafka通過批次的概念來 提高吞吐量，但是也會在增加延遲。
? ? ? ? ?*/
? ? ? ? // 以下配置當(dāng)緩存數(shù)量達(dá)到16kb，就會觸發(fā)網(wǎng)絡(luò)請求，發(fā)送消息
// ? ? ? ?props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
? ? ? ? // 每條消息在緩存中的最長時間，如果超過這個時間就會忽略batch.size的限制，由客戶端立即將消息發(fā)送出去
// ? ? ? ?props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// ? ? ? ?props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
? ? ? ? props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? return props;
? ? }
? ??
? ? @Value("${spring.outkafka.bootstrap-servers}")
? ? private String outServers;
? ? @Value("${spring.outkafka.consumer.group-id}")
? ? private String outGroupid;
? ? @Value("${spring.outkafka.consumer.enable-auto-commit}")
? ? private String outEnableAutoCommit;
? ??
?
? ? static {
? ? ? ??
? ? }
? ??
? ? /**
? ? ?* 連接第二個kafka集群的配置
? ? ?*/
? ? @Bean
? ? KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactoryOutSchedule() {
? ? ? ? ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
? ? ? ? factory.setConsumerFactory(consumerFactoryOutSchedule());
? ? ? ? factory.setConcurrency(3);
? ? ? ? factory.getContainerProperties().setPollTimeout(3000);
? ? ? ? return factory;
? ? }
?
? ? @Bean
? ? public ConsumerFactory<Integer, String> consumerFactoryOutSchedule() {
? ? ? ? return new DefaultKafkaConsumerFactory<>(consumerConfigsOutSchedule());
? ? }
?
? ? /**
? ? ?* 連接第二個集群的消費者配置
? ? ?*/
? ? @Bean
? ? public Map<String, Object> consumerConfigsOutSchedule() {
? ? ? ? Map<String, Object> props = new HashMap<>();
? ? ? ? props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
? ? ? ? props.put(ConsumerConfig.GROUP_ID_CONFIG, outGroupid);
? ? ? ? props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, outEnableAutoCommit);
? ? ? ? props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? return props;
? ? }
? ??
? ? @Bean //生產(chǎn)者工廠配置
? ? public ProducerFactory<String, String> producerOutFactory() {
? ? ? ? return new DefaultKafkaProducerFactory<>(senderOutProps());
? ? }
? ??
? ? @Bean //kafka發(fā)送消息模板
? ? public KafkaTemplate<String, String> kafkaOutTemplate() {
? ? ? ? return new KafkaTemplate<String, String>(producerOutFactory());
? ? }
? ??
? ? /**
? ? ?* 生產(chǎn)者配置方法
? ? ?*
? ? ?* 生產(chǎn)者有三個必選屬性
? ? ?* <p>
? ? ?* 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
? ? ?* 生產(chǎn)者會從給定的broker里查找到其他broker的信息。不過建議至少提供兩個broker信息，一旦 其中一個宕機(jī)，生產(chǎn)者仍能能夠連接到集群上。
? ? ?* </p>
? ? ?* <p>
? ? ?* 2.key.serializer broker希望接收到的消息的鍵和值都是字節(jié)數(shù)組。 生產(chǎn)者用對應(yīng)的類把鍵對象序列化成字節(jié)數(shù)組。
? ? ?* </p>
? ? ?* <p>
? ? ?* 3.value.serializer 值得序列化方式
? ? ?* </p>
? ? ?*
? ? ?*
? ? ?* @return
? ? ?*/
? ? private Map<String, Object> senderOutProps() {
? ? ? ? Map<String, Object> props = new HashMap<>();
? ? ? ? props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
? ? ? ? /**
? ? ? ? ?* 當(dāng)從broker接收到的是臨時可恢復(fù)的異常時，生產(chǎn)者會向broker重發(fā)消息，但是不能無限
? ? ? ? ?* 制重發(fā)，如果重發(fā)次數(shù)達(dá)到限制值，生產(chǎn)者將不會重試并返回錯誤。
? ? ? ? ?* 通過retries屬性設(shè)置。默認(rèn)情況下生產(chǎn)者會在重試后等待100ms，可以通過 retries.backoff.ms屬性進(jìn)行修改
? ? ? ? ?*/
? ? ? ? props.put(ProducerConfig.RETRIES_CONFIG, 0);
? ? ? ? /**
? ? ? ? ?* 在考慮完成請求之前，生產(chǎn)者要求leader收到的確認(rèn)數(shù)量。這可以控制發(fā)送記錄的持久性。允許以下設(shè)置：
? ? ? ? ?* <ul>
? ? ? ? ?* <li>
? ? ? ? ?* <code> acks = 0 </ code>如果設(shè)置為零，則生產(chǎn)者將不會等待來自服務(wù)器的任何確認(rèn)。該記錄將立即添加到套接字緩沖區(qū)并視為已發(fā)送。在這種情況下，無法保證服務(wù)器已收到記錄，并且
? ? ? ? ?* <code>retries </ code>配置將不會生效（因為客戶端通常不會知道任何故障）。為每條記錄返回的偏移量始終設(shè)置為-1。
? ? ? ? ?* <li> <code> acks = 1 </code>
? ? ? ? ?* 這意味著leader會將記錄寫入其本地日志，但無需等待所有follower的完全確認(rèn)即可做出回應(yīng)。在這種情況下，
? ? ? ? ?* 如果leader在確認(rèn)記錄后立即失敗但在關(guān)注者復(fù)制之前，則記錄將丟失。
? ? ? ? ?* <li><code> acks = all </code>
? ? ? ? ?* 這意味著leader將等待完整的同步副本集以確認(rèn)記錄。這保證了只要至少一個同步副本仍然存活，記錄就不會丟失。這是最強(qiáng)有力的保證。
? ? ? ? ?* 這相當(dāng)于acks = -1設(shè)置
? ? ? ? ?*/
? ? ? ? props.put(ProducerConfig.ACKS_CONFIG, "1");
? ? ? ? /**
? ? ? ? ?* 當(dāng)有多條消息要被發(fā)送到統(tǒng)一分區(qū)是，生產(chǎn)者會把他們放到統(tǒng)一批里。kafka通過批次的概念來 提高吞吐量，但是也會在增加延遲。
? ? ? ? ?*/
? ? ? ? // 以下配置當(dāng)緩存數(shù)量達(dá)到16kb，就會觸發(fā)網(wǎng)絡(luò)請求，發(fā)送消息
// ? ? ? ?props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
? ? ? ? // 每條消息在緩存中的最長時間，如果超過這個時間就會忽略batch.size的限制，由客戶端立即將消息發(fā)送出去
// ? ? ? ?props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// ? ? ? ?props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
? ? ? ? props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? return props;
? ? }
}發(fā)送工具類MyKafkaProducer
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
?
import lombok.extern.slf4j.Slf4j;
?
/**
?* <p>
?* <b>KafkaProducer Description:</b> kafka生產(chǎn)者
?* </p>
?*
?* @author douzaixing<b>DATE</b> 2019年7月8日 下午4:09:29
?*/
@Component // 這個必須加入容器不然，不會執(zhí)行
@EnableScheduling // 這里是為了測試加入定時調(diào)度
@Slf4j
public class MyKafkaProducer {
?
? ? @Autowired
? ? private KafkaTemplate<String, String> kafkaTemplate;
?
? ? @Autowired
? ? private KafkaTemplate<String, String> kafkaOutTemplate;
?
? ? public ListenableFuture<SendResult<String, String>> send(String topic, String key, String json) {
? ? ? ? ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, key, json);
? ? ? ? log.info("inner kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功===========");
? ? ? ? return result;
? ? }
?
? ? public ListenableFuture<SendResult<String, String>> sendOut(String topic, String key, String json) {
? ? ? ? ListenableFuture<SendResult<String, String>> result = kafkaOutTemplate.send(topic, key, json);
? ? ? ? log.info("out kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功===========");
? ? ? ? return result;
? ? }
?
}測試類
@Slf4j
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes={OesBcServiceApplication.class})
public class MoreKafkaTest {
? ??
? ? @Autowired
? ? private MyKafkaProducer kafkaProducer;
? ??
? ? @Test
? ? public void sendInner() {
? ? ? ? for (int i = 0; i < 1; i++) {
? ? ? ? ? ? kafkaProducer.send("inner_test", "douzi" + i, "liyuehua" + i);
? ? ? ? ? ? kafkaProducer.sendOut("out_test", "douziout" + i, "fanbingbing" + i);
? ? ? ? }
? ? }
}接收類
@Component
@Slf4j
public class KafkaConsumer { ?
? ? @KafkaListener(topics={"inner_test"}, containerFactory="kafkaListenerContainerFactory")
? ? public void innerlistener(ConsumerRecord<String, String> record) {
? ? ? ? log.info("inner kafka receive #key=" + record.key() + "#value=" + record.value());
? ? }
? ??
? ? @KafkaListener(topics={"out_test"}, containerFactory="kafkaListenerContainerFactoryOutSchedule")
? ? public void outListener(ConsumerRecord<String, String> record) {
? ? ? ? log.info("out kafka receive #key=" + record.key() + "#value=" + record.value());
? ? }
}測試結(jié)果
07-11 12:41:27.811 INFO [com.wondertek.oes.bc.service.send.MyKafkaProducer] - inner kafka send #topic=inner_test#key=douzi0#json=liyuehua0#推送成功===========
07-11 12:41:27.995 INFO [com.wondertek.oes.bc.service.send.KafkaConsumer] - inner kafka receive #key=douzi0#value=liyuehua0
07-11 12:41:28.005 INFO [com.wondertek.oes.bc.service.send.MyKafkaProducer] - out kafka send #topic=out_test#key=douziout0#json=fanbingbing0#推送成功===========
07-11 12:41:28.013 INFO [com.wondertek.oes.bc.service.send.KafkaConsumer] - out kafka receive #key=douziout0#value=fanbingbing0
總結(jié)
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了MyEclipse+Tomcat配置詳解(圖文)，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2019-01-01
這篇文章主要介紹了MyBatis中的limit分頁設(shè)置方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
這篇文章主要介紹了Java反射機(jī)制深入理解的相關(guān)資料,希望通過本文大家能理解這部分內(nèi)容，需要的朋友可以參考下2017-09-09
Java中的線程同步與ThreadLocal無鎖化線程封閉實現(xiàn)
這篇文章主要介紹了Java中的線程同步與ThreadLocal無鎖化線程封閉實現(xiàn),Synchronized關(guān)鍵字與ThreadLocal變量的使用是Java中線程控制的基礎(chǔ),需要的朋友可以參考下2016-03-03
從java源碼分析線程池(池化技術(shù))的實現(xiàn)原理
這篇文章主要介紹了從java源碼分析線程池(池化技術(shù))的實現(xiàn)原理，池化技術(shù)是一種編程技巧，當(dāng)程序出現(xiàn)高并發(fā)時，能夠明顯的優(yōu)化程序，降低系統(tǒng)頻繁創(chuàng)建銷毀連接等額外開銷，下文更多的相關(guān)介紹需要的小伙伴可以參考一下2022-04-04
本文詳細(xì)介紹了SpringBoot的日志功能,包括默認(rèn)日志框架Logback,日志級別配置,日志格式自定義,日志文件輸出,日志歸檔與清理,自定義日志配置,與其他日志框架的集成以及日志性能優(yōu)化,通過結(jié)合實際場景,提供了詳細(xì)的配置與實踐指南,感興趣的朋友一起看看吧2025-01-01
這篇文章主要介紹了SpringMVC中的Model對象用法說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06
這篇文章主要介紹了Spring整合redis的操作代碼，本文通過實例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定參考借鑒價值,需要的朋友可以參考下2022-02-02
這篇文章主要介紹了新手場景Java線程相關(guān)問題及解決方案,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-07-07
這篇文章主要介紹了深入理解java泛型Generic,文中有非常詳細(xì)的代碼示例,對正在學(xué)習(xí)java的小伙伴們有非常好的幫助,需要的朋友可以參考下2021-05-05