springboot怎么配置雙kafka

這篇文章主要介紹“springboot怎么配置雙kafka”，在日常操作中，相信很多人在springboot怎么配置雙kafka問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”springboot怎么配置雙kafka”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

springboot配置雙kafka

使用spring boot 2.0.8.RELEASE 版本

引入Maven kafka jar、準(zhǔn)備兩個(gè)kafka；

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

配置yml配置文件

spring:
  kafka:
    bootstrap-servers: 180.167.180.242:9092 #kafka的訪問(wèn)地址，多個(gè)用","隔開(kāi)
    consumer:
      enable-auto-commit: true
      group-id: kafka #群組ID
  outkafka:
    bootstrap-servers: localhost:9092 #kafka的訪問(wèn)地址，多個(gè)用","隔開(kāi)
    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)前容器下的消費(fèi)者工廠
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
 
    @Bean//第一個(gè)消費(fèi)者工廠的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)者有三個(gè)必選屬性
     * <p>
     * 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
     * 生產(chǎn)者會(huì)從給定的broker里查找到其他broker的信息。不過(guò)建議至少提供兩個(gè)broker信息，一旦 其中一個(gè)宕機(jī)，生產(chǎn)者仍能能夠連接到集群上。
     * </p>
     * <p>
     * 2.key.serializer broker希望接收到的消息的鍵和值都是字節(jié)數(shù)組。 生產(chǎn)者用對(duì)應(yīng)的類把鍵對(duì)象序列化成字節(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接收到的是臨時(shí)可恢復(fù)的異常時(shí)，生產(chǎn)者會(huì)向broker重發(fā)消息，但是不能無(wú)限
         * 制重發(fā)，如果重發(fā)次數(shù)達(dá)到限制值，生產(chǎn)者將不會(huì)重試并返回錯(cuò)誤。
         * 通過(guò)retries屬性設(shè)置。默認(rèn)情況下生產(chǎn)者會(huì)在重試后等待100ms，可以通過(guò) retries.backoff.ms屬性進(jìn)行修改
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        /**
         * 在考慮完成請(qǐng)求之前，生產(chǎn)者要求leader收到的確認(rèn)數(shù)量。這可以控制發(fā)送記錄的持久性。允許以下設(shè)置：
         * <ul>
         * <li>
         * <code> acks = 0 </ code>如果設(shè)置為零，則生產(chǎn)者將不會(huì)等待來(lái)自服務(wù)器的任何確認(rèn)。該記錄將立即添加到套接字緩沖區(qū)并視為已發(fā)送。在這種情況下，無(wú)法保證服務(wù)器已收到記錄，并且
         * <code>retries </ code>配置將不會(huì)生效（因?yàn)榭蛻舳送ǔ２粫?huì)知道任何故障）。為每條記錄返回的偏移量始終設(shè)置為-1。
         * <li> <code> acks = 1 </code>
         * 這意味著leader會(huì)將記錄寫(xiě)入其本地日志，但無(wú)需等待所有follower的完全確認(rèn)即可做出回應(yīng)。在這種情況下，
         * 如果leader在確認(rèn)記錄后立即失敗但在關(guān)注者復(fù)制之前，則記錄將丟失。
         * <li><code> acks = all </code>
         * 這意味著leader將等待完整的同步副本集以確認(rèn)記錄。這保證了只要至少一個(gè)同步副本仍然存活，記錄就不會(huì)丟失。這是最強(qiáng)有力的保證。
         * 這相當(dāng)于acks = -1設(shè)置
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        /**
         * 當(dāng)有多條消息要被發(fā)送到統(tǒng)一分區(qū)是，生產(chǎn)者會(huì)把他們放到統(tǒng)一批里。kafka通過(guò)批次的概念來(lái) 提高吞吐量，但是也會(huì)在增加延遲。
         */
        // 以下配置當(dāng)緩存數(shù)量達(dá)到16kb，就會(huì)觸發(fā)網(wǎng)絡(luò)請(qǐng)求，發(fā)送消息
//        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 每條消息在緩存中的最長(zhǎng)時(shí)間，如果超過(guò)這個(gè)時(shí)間就會(huì)忽略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 {
        
    }
    
    /**
     * 連接第二個(gè)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());
    }
 
    /**
     * 連接第二個(gè)集群的消費(fèi)者配置
     */
    @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)者有三個(gè)必選屬性
     * <p>
     * 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
     * 生產(chǎn)者會(huì)從給定的broker里查找到其他broker的信息。不過(guò)建議至少提供兩個(gè)broker信息，一旦 其中一個(gè)宕機(jī)，生產(chǎn)者仍能能夠連接到集群上。
     * </p>
     * <p>
     * 2.key.serializer broker希望接收到的消息的鍵和值都是字節(jié)數(shù)組。 生產(chǎn)者用對(duì)應(yīng)的類把鍵對(duì)象序列化成字節(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接收到的是臨時(shí)可恢復(fù)的異常時(shí)，生產(chǎn)者會(huì)向broker重發(fā)消息，但是不能無(wú)限
         * 制重發(fā)，如果重發(fā)次數(shù)達(dá)到限制值，生產(chǎn)者將不會(huì)重試并返回錯(cuò)誤。
         * 通過(guò)retries屬性設(shè)置。默認(rèn)情況下生產(chǎn)者會(huì)在重試后等待100ms，可以通過(guò) retries.backoff.ms屬性進(jìn)行修改
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        /**
         * 在考慮完成請(qǐng)求之前，生產(chǎn)者要求leader收到的確認(rèn)數(shù)量。這可以控制發(fā)送記錄的持久性。允許以下設(shè)置：
         * <ul>
         * <li>
         * <code> acks = 0 </ code>如果設(shè)置為零，則生產(chǎn)者將不會(huì)等待來(lái)自服務(wù)器的任何確認(rèn)。該記錄將立即添加到套接字緩沖區(qū)并視為已發(fā)送。在這種情況下，無(wú)法保證服務(wù)器已收到記錄，并且
         * <code>retries </ code>配置將不會(huì)生效（因?yàn)榭蛻舳送ǔ２粫?huì)知道任何故障）。為每條記錄返回的偏移量始終設(shè)置為-1。
         * <li> <code> acks = 1 </code>
         * 這意味著leader會(huì)將記錄寫(xiě)入其本地日志，但無(wú)需等待所有follower的完全確認(rèn)即可做出回應(yīng)。在這種情況下，
         * 如果leader在確認(rèn)記錄后立即失敗但在關(guān)注者復(fù)制之前，則記錄將丟失。
         * <li><code> acks = all </code>
         * 這意味著leader將等待完整的同步副本集以確認(rèn)記錄。這保證了只要至少一個(gè)同步副本仍然存活，記錄就不會(huì)丟失。這是最強(qiáng)有力的保證。
         * 這相當(dāng)于acks = -1設(shè)置
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        /**
         * 當(dāng)有多條消息要被發(fā)送到統(tǒng)一分區(qū)是，生產(chǎn)者會(huì)把他們放到統(tǒng)一批里。kafka通過(guò)批次的概念來(lái) 提高吞吐量，但是也會(huì)在增加延遲。
         */
        // 以下配置當(dāng)緩存數(shù)量達(dá)到16kb，就會(huì)觸發(fā)網(wǎng)絡(luò)請(qǐng)求，發(fā)送消息
//        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 每條消息在緩存中的最長(zhǎng)時(shí)間，如果超過(guò)這個(gè)時(shí)間就會(huì)忽略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 // 這個(gè)必須加入容器不然，不會(huì)執(zhí)行
@EnableScheduling // 這里是為了測(cè)試加入定時(shí)調(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;
    }
 
}

測(cè)試類

@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());
    }
}

測(cè)試結(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

到此，關(guān)于“springboot怎么配置雙kafka”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！