SpringBoot怎么集成Kafka配置工具類

本篇內容介紹了“SpringBoot怎么集成Kafka配置工具類”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

spring-kafka 是基于 java版的 kafka client與spring的集成，提供了 KafkaTemplate，封裝了各種方法，方便操作,它封裝了apache的kafka-client,不需要再導入client依賴

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

YML配置

kafka:
    #bootstrap-servers: server1:9092,server2:9093 #kafka開發地址,
    #生產者配置
    producer:
      # Kafka提供的序列化和反序列化類
      key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 1 # 消息發送重試次數
      #acks = 0：設置成 表示 producer 完全不理睬 leader broker 端的處理結果。此時producer 發送消息后立即開啟下 條消息的發送，根本不等待 leader broker 端返回結果
      #acks= all 或者－1 ：表示當發送消息時， leader broker 不僅會將消息寫入本地日志，同時還會等待所有其他副本都成功寫入它們各自的本地日志后，才發送響應結果給，消息安全但是吞吐量會比較低。
      #acks = 1：默認的參數值。 producer 發送消息后 leader broker 僅將該消息寫入本地日志，然后便發送響應結果給producer ，而無須等待其他副本寫入該消息。折中方案，只要leader一直活著消息就不會丟失，同時也保證了吞吐量
      acks: 1 #應答級別:多少個分區副本備份完成時向生產者發送ack確認(可選0、1、all/-1)
      batch-size: 16384 #批量大小
      properties:
        linger:
          ms: 0 #提交延遲
      buffer-memory: 33554432 # 生產端緩沖區大小
    # 消費者配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 分組名稱
      group-id: web
      enable-auto-commit: false
      #提交offset延時(接收到消息后多久提交offset)
      # auto-commit-interval: 1000ms
      #當kafka中沒有初始offset或offset超出范圍時將自動重置offset
      # earliest:重置為分區中最小的offset;
      # latest:重置為分區中最新的offset(消費分區中新產生的數據);
      # none:只要有一個分區不存在已提交的offset,就拋出異常;
      auto-offset-reset: latest
      properties:
        #消費會話超時時間(超過這個時間consumer沒有發送心跳,就會觸發rebalance操作)
        session.timeout.ms: 15000
        #消費請求超時時間
        request.timeout.ms: 18000
      #批量消費每次最多消費多少條消息
      #每次拉取一條，一條條消費，當然是具體業務狀況設置
      max-poll-records: 1
      # 指定心跳包發送頻率，即間隔多長時間發送一次心跳包，優化該值的設置可以減少Rebalance操作，默認時間為3秒；
      heartbeat-interval: 6000
      # 發出請求時傳遞給服務器的 ID。用于服務器端日志記錄 正常使用后解開注釋，不然只有一個節點會報錯
      #client-id: mqtt
    listener:
      #消費端監聽的topic不存在時，項目啟動會報錯(關掉)
      missing-topics-fatal: false
      #設置消費類型 批量消費 batch，單條消費：single
      type: single
      #指定容器的線程數，提高并發量
      #concurrency: 3
      #手動提交偏移量 manual達到一定數據后批量提交
      #ack-mode: manual
      ack-mode: MANUAL_IMMEDIATE #手動確認消息
        # 認證
    #properties:
      #security:
        #protocol: SASL_PLAINTEXT
      #sasl:
        #mechanism: SCRAM-SHA-256
        #jaas:config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'

簡單工具類，能滿足正常使用，主題是無法修改的

@Component
@Slf4j
public class KafkaUtils<K, V> {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] servers;

    /**
     * 獲取連接
     * @return
     */
    private Admin getAdmin() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        // 正式環境需要添加賬號密碼
        return Admin.create(properties);
    }

    /**
     * 增加topic
     *
     * @param name      主題名字
     * @param partition 分區數量
     * @param replica   副本數量
     * @date 2022-06-23 chens
     */
    public R addTopic(String name, Integer partition, Integer replica) {
        Admin admin = getAdmin();
        if (replica > servers.length) {
            return R.error("副本數量不允許超過Broker數量");
        }
        try {
            NewTopic topic = new NewTopic(name, partition, Short.parseShort(replica.toString()));
            admin.createTopics(Collections.singleton(topic));
        } finally {
            admin.close();
        }
        return R.ok();
    }

    /**
     * 刪除主題
     *
     * @param names 主題名字集合
     * @date 2022-06-23 chens
     */
    public void deleteTopic(List<String> names) {
        Admin admin = getAdmin();
        try {
            admin.deleteTopics(names);
        } finally {
            admin.close();
        }
    }

    /**
     * 查詢所有主題
     *
     * @date 2022-06-24 chens
     */
    public Set<String> queryTopic() {
        Admin admin = getAdmin();
        try {
            ListTopicsResult topics = admin.listTopics();
            Set<String> set = topics.names().get();
            return set;
        } catch (Exception e) {
            log.error("查詢主題錯誤！");
        } finally {
            admin.close();
        }
        return null;
    }

    // 向所有分區發送消息
    public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
        return kafkaTemplate.send(topic, data);
    }
    
    // 指定key發送消息，相同key保證消息在同一個分區
    public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, key, data);
    }

    // 指定分區和key發送。
    public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, partition, key, data);
    }
}

發送消息 使用異步

@GetMapping("/{topic}")
    public String test(@PathVariable String topic, @PathVariable Long index) throws ExecutionException, InterruptedException {

        ListenableFuture future = null;
        Chenshuang user = new Chenshuang(i, "陳爽", "123456", new Date());
        String s = JSON.toJSONString(user);
        KafkaUtils utils = new KafkaUtils();
        future = kafkaUtils.send(topic, s);
        // 異步回調，同步get,會等待 不推薦同步！
        future.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("發送失敗");
            }
            @Override
            public void onSuccess(Object result) {
                System.out.println("發送成功：" + result);
            }
        });
        return "發送成功";
    }

建立主題

如果broker端配置auto.create.topics.enable為true(默認為true)，當收到客戶端的元數據請求時則會創建topic。

向一個不存在的主題發送和消費都會創建一個新的主題，很多時候，非預期的創建主題，會導致很多意想不到的問題，建議關掉該特性。

Topic主題用來區分不同類型的消息，實際也就是適用于不同的業務場景，默認消息保存一周時間；

同一個Topic主題下，默認是一個partition分區，也就是只能有一個消費者來消費，如果想提升消費能力，就需要增加分區；

同一個Topic的多個分區，可以有三種方式分派消息(key,value)到不同的分區，指定分區、HASH路由、默認，同一個分區內的消息ID唯一，并順序；

消費者消費partition分區內的消息時，是通過offsert來標識消息的位置；

GroupId用來解決同一個Topic主題下重復消費問題，比如一條消費需要多個消費者接收到，就可以通過設置不同的GroupId實現，

實際消息是存一份的，只是通過邏輯上設置標識來區分，系統會記錄Topic主題下–》GroupId分組下–》partition分區下的offsert，來標識是否消費過。

發送消息的高可用—

集群模式，多副本方式實現；一條消息的提交，可能通過設置acks標識實現不同的可用性，=0時，發送成功就OK；=1時，master成功響應才OK，=all時，一半以上的響應才OK(真正的高可用)

消費消息的高可用—

可以關閉自動標識offsert模式，先拉取消息，消費完成后，再去設置offsert位置，來解決消費高可用

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaTopic {
    // yml自定義主題，項目啟動就創建，
    @Value("${spring.kafka.topic}")
    String topic;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] server;
    /**
     * 項目啟動 初始化主題,如果存在不會覆蓋主題的
     */
    @Bean
    public NewTopic batchTopic() {
        // 最大復制因子 <= 經紀人broker數量.
        return new NewTopic(topic, 10, (short) server.length);
    }
}

監聽類 ,一條消息，各分組內的消費者只有一個消費者消費一次，如果消息在1區，指定分區1監聽也會消費
也可以同個方法監聽不同的主題，指定位移監聽
同組會均勻消費，不同組會重復消費。

1、單播模式，只有一個消費者組

（1）topic只有1個partition,該組內有多個消費者時，此時同一個partition內的消息只能被該組中的一 個consumer消費。當消費者數量多于partition數量時，多余的消費者是處于空閑狀態的,如圖1所示。topic，test只有一個partition，并且只有1個group，G1,該group內有多個consumer,只能被其中一個消費者消費，其他的處于空閑狀態。

圖一

（2）該topic有多個partition,該組內有多個消費者，比如test 有3個partition,該組內有2個消費者，那么可能就是C0對應消費p0，p1內的數據，c1對應消費p2的數據；如果有3個消費者，就是一個消費者對應消費一個partition內的數據了。圖解分別如圖2，圖3.這種模式在集群模式下使用是非常普遍的，比如我們可以起3個服務，對應的topic設置3個partiition，這樣就可以實現并行消費，大大提高處理消息的效率。

圖二

圖三

2、廣播模式，多個消費者組

如果想實現廣播的模式就需要設置多個消費者組，這樣當一個消費者組消費完這個消息后，絲毫不影響其他組內的消費者進行消費，這就是廣播的概念。

（1）多個消費者組，1個partition

該topic內的數據被多個消費者組同時消費，當某個消費者組有多個消費者時也只能被一個消費者消費，如圖4所示：

圖四

（2）多個消費者組，多個partition

該topic內的數據可被多個消費者組多次消費，在一個消費者組內，每個消費者又可對應該topic內的一個或者多個partition并行消費，如圖五：

注意： 消費者的數量并不能決定一個topic的并行度。它是由分區的數目決定的。
再多的消費者，分區數少，也是浪費！
一個組的最大并行度將等于該主題的分區數。

@Component
@Slf4j
public class Consumer {
    // 監聽主題 分組a
    @KafkaListener(topics =("${spring.kafka.topic}") ,groupId = "a")
    public  void  getMessage(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    // 監聽主題 分組a
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "a")
    public  void getMessage2(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    // 監聽主題 分組b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage3(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }
    // 監聽主題 分組b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage4(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }

    // 指定監聽分區1的消息
    @KafkaListener(topicPartitions = {@TopicPartition(topic = ("${spring.kafka.topic}"),partitions = {"1"})})
    public void getMessage5(ConsumerRecord message, Acknowledgment ack) {
        Long id = JSONObject.parseObject(message.value().toString()).getLong("id");
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }
    
    /**
     * @Title 指定topic、partition、offset消費
     * @Description 同時監聽topic1和topic2，監聽topic1的0號分區、topic2的 "0號和1號" 分區，指向1號分區的offset初始值為8
     * 注意：topics和topicPartitions不能同時使用；
     **/
    @KafkaListener(id = "c1",groupId = "c",topicPartitions = {
            @TopicPartition(topic = "t1", partitions = { "0" }),
            @TopicPartition(topic = "t2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))})
    public void getMessage6(ConsumerRecord record,Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    
    /**        
     * 批量消費監聽goods變更消息
     * yml配置listener:type 要改為batch
     * ymk配置consumer：max-poll-records: ？？（每次拉取多少條數據消費）
     * concurrency = "2" 啟動多少線程執行，應小于等于broker數量，避免資源浪費
     */
    @KafkaListener(id="sync-modify-goods", topics = "${spring.kafka.topic}",concurrency = "4")
    public void getMessage7(List<ConsumerRecord<String, String>> records){
        for (ConsumerRecord<String, String> msg:records) {
            GoodsChangeMsg changeMsg = null;
            try {
                changeMsg = JSONObject.parseObject(msg.value(), GoodsChangeMsg.class);
                syncGoodsProcessor.handle(changeMsg);
            }catch (Exception exception) {
                log.error("解析失敗{}", msg, exception);
            }
        }
    }
}

“SpringBoot怎么集成Kafka配置工具類”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！