流式圖表拒絕增刪改查之kafka核心消費(fèi)邏輯下篇

更新時(shí)間：2023年04月12日 15:18:59 作者：在下uptown

這篇文章主要為大家介紹了流式圖表拒絕增刪改查之kafka核心消費(fèi)邏輯講解的下篇，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前篇回顧

kafka消費(fèi)者線程

突擊檢查八股文，實(shí)現(xiàn)線程的方法有哪些？嗯？沒復(fù)習(xí)是吧，行沒關(guān)系，那感謝參加本次面試哈。

常用的幾種方式分別是：

繼承Thread類，重寫run方法
實(shí)現(xiàn)Runbale接口，重寫run方法
實(shí)現(xiàn)Callable接口，重寫call方法

這里我們直接創(chuàng)捷出一個(gè)任務(wù)類實(shí)現(xiàn)Runable方法，重寫run方法，一個(gè)線程當(dāng)作一個(gè)kafka client，所以要在任務(wù)類中聲明一個(gè)KafkaConsumer的成員變量，另外創(chuàng)建任務(wù)需要指定當(dāng)前任務(wù)的名稱也就是線程名，還有要監(jiān)聽的topic主題。

private KafkaConsumer<String, String> consumer;
private String topic;
private String threadName;

name和topic通過構(gòu)造方法傳進(jìn)來，同時(shí)在構(gòu)造方法里完成對(duì)client的初始化操作。

/**
    * 封裝必要信息
    * @param bootServer 生產(chǎn)者ip
    * @param groupId 分組信息
    * @param topic  訂閱主題
    */
   public KafkaConsumerRunnable(String bootServer, String groupId, String topic) {
       this.topic = topic;
       Properties props = new Properties();
       props.put("bootstrap.servers", bootServer);
       props.put("group.id", groupId);
       props.put("enable.auto.commit", "false");
       props.put("auto.offset.reset", "latest");
       props.put("max.poll.records", 5);
       props.put("session.timeout.ms", "60000");
       props.put("max.poll.interval.ms", 300000);
       props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //鍵反序列化方式
       props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       this.consumer = new KafkaConsumer&lt;&gt;(props);
   }

這里封裝kafka client的必要信息，入?yún)ootServer為kafka集群ip，groupId為threadName，我們規(guī)定一個(gè)線程為一個(gè)kafka消費(fèi)鏈接，消費(fèi)一個(gè)topic。

上一篇線程池保證了任務(wù)不會(huì)輕易掛掉，就算掛掉了也會(huì)重新提交，所以為了節(jié)省資源不做所謂的同groupId的負(fù)載操作。session.timeout.ms和max.poll.interval.ms可以根據(jù)當(dāng)前的kafka資源靈活配置，不然可能會(huì)引發(fā)一些reblance。

enable.auto.commit設(shè)置為false，手動(dòng)提交offset，auto.offset.reset這塊由于業(yè)務(wù)特殊，本來就是流式圖表瞬時(shí)的展示，如果真的出現(xiàn)了數(shù)據(jù)丟失那就丟了吧，從最新的數(shù)據(jù)讀取。

接下來只需要處理下消費(fèi)邏輯，consumer.subscribe(Collections.singletonList(this.topic))開始訂閱監(jiān)聽kafka數(shù)據(jù)，搞一個(gè)while true不斷的消費(fèi)數(shù)據(jù)，try catch只需要對(duì)WakeupException做處理，kafka客戶端會(huì)在關(guān)閉的時(shí)候拋出WakeupException異常。

finally里提交offset，無論這條offset對(duì)應(yīng)的數(shù)據(jù)消費(fèi)成功還是失敗都是消費(fèi)過了，失敗了就過去了。

   @Override
   public void run() {
   consumer.subscribe(Collections.singletonList(this.topic));
   String key = "stream_chart:" + this.name;
   Thread.currentThread().setName(key);
   try {
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
         // 如果隊(duì)列中沒有消息 等待KAFKA_TIME_OUT后調(diào)用poll，如果有消息立即消費(fèi)
         for (ConsumerRecord<String, String> record : records) {
            String value = record.value();
            log.info("線程 {} 消費(fèi)kafka數(shù)據(jù) -> {} \n", Thread.currentThread().getName(), value);
            RedisConfig.getRedisTemplate().opsForZSet().add(key, value, Instant.now().getEpochSecond() * 1000);
         }
      }
   } catch (WakeupException e) {
      log.info("ignore for shutdown", e);
   } finally {
      consumer.commitAsync();
   }
}

我們消費(fèi)到數(shù)據(jù)直接放到redis的zset結(jié)構(gòu)里，當(dāng)前的時(shí)間戳作為score，最后留一個(gè)關(guān)閉客戶端的后門

// 退出后關(guān)掉客戶端
public void shutDown() {
   consumer.wakeup();
}

任務(wù)提交

任務(wù)提交這塊只需要在業(yè)務(wù)service中注入線程池，創(chuàng)建對(duì)應(yīng)的KafkaRunable任務(wù)封裝對(duì)應(yīng)的信息，執(zhí)行execute即可。

這里有個(gè)坑需要注意下，第二次突擊檢查八股文，線程池提交方法submit與execute的區(qū)別說一下。不知道的立刻去熟讀并背誦。

public class TestTheadPool {
    public static void main(String[] args) {
        ExecutorService executorService= Executors.newFixedThreadPool(1);
        executorService.submit(new task("submit"));
        executorService.execute(new task("execute"));
    }
}
class task implements  Runnable{
    private String name;
    public task(String name) {
        this.name = name;
    }
    @Override
    public void run() {
        System.out.println(this.name + " start task");
        int i=1/0;
    }
}

熟悉的同學(xué)通過示例代碼可以看出來，submit提交的線程不會(huì)拋出異常代碼，只有獲取Future返回值并執(zhí)行g(shù)et方法才會(huì)捕獲到異常。這塊涉及到異步的東西不再贅述

try {
    Future<?> submit = executorService.submit(new task("submit"));
    submit.get();
} catch (InterruptedException e) {
    e.printStackTrace();
} catch (ExecutionException e) {
    e.printStackTrace();
}

所以我們要使用execute執(zhí)行，不然kafka消費(fèi)線程里消費(fèi)失敗了攔截不到就不會(huì)被重新提交，導(dǎo)致線程掛掉。

以上就是流式圖表拒絕增刪改查之kafka核心消費(fèi)邏輯下篇的詳細(xì)內(nèi)容，更多關(guān)于kafka消費(fèi)流式圖表的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: