kafka集群

image.png

Kafka的設(shè)計(jì)都是為了實(shí)現(xiàn)kafak消息隊(duì)列消費(fèi)數(shù)據(jù)的語(yǔ)義
Kafka消息隊(duì)列中數(shù)據(jù)消費(fèi)的三種語(yǔ)義：

• at-most-once:至多一次
  會(huì)出現(xiàn)數(shù)據(jù)丟失的問題
• at-least-once:至少一次
  會(huì)出現(xiàn)數(shù)據(jù)重復(fù)的問題
• exactly-once:有且僅有一次
  只消費(fèi)處理成功一次
  所有消息隊(duì)列的目標(biāo)

kakfa卡只能保證局部有序,如何實(shí)現(xiàn)全局有序

image.png

kafka中每個(gè)分區(qū)內(nèi)有offset保證局部有序
1.如果topic只有 一個(gè)分區(qū)可以實(shí)現(xiàn)全局有序
2.寫數(shù)據(jù)的時(shí)候指定分區(qū)編號(hào),只想一個(gè)分區(qū)內(nèi)寫數(shù)據(jù)

Kafka如何保證自身數(shù)據(jù)安全

多副本機(jī)制
生產(chǎn)者和消費(fèi)者跟leader副本進(jìn)行讀寫
follower副本跟leader副本同步數(shù)據(jù)
且讀寫的內(nèi)容只能在hw高水位線之前

副本同步機(jī)制
為什么會(huì)存在同步？因?yàn)樽x寫只跟leader副本打交道。所以follower副本需要同步數(shù)據(jù)。
副本集合概念:

• AR 該分區(qū)的所有副本 ALL

• ISR 所有副本中處在同步狀態(tài)的 健康良好的 IN Sync
  只有ISR中的副本才有選舉資格 成為新的leader 優(yōu)先順序 從左往右

  誰(shuí)來(lái)負(fù)責(zé)leader副本選舉？ kafka中主角色controller
  誰(shuí)來(lái)kafka主角色controller選舉呢？ 通過zk集群選舉。【臨時(shí)節(jié)點(diǎn) 監(jiān)聽 唯一性】

• OSR 非同步狀態(tài)的 Out Sync
  正常情況下 OSR集合中應(yīng)該是空的 每個(gè)副本都應(yīng)該處于健康的狀態(tài)
  ISR同步超時(shí) 就會(huì)被打入OSR

        AR=ISR+OSR
        
    名詞：
        LSO  start  每個(gè)副本的第一個(gè)消息offset 正常是0。如果開啟了刪除清理 之前的數(shù)據(jù)就會(huì)被刪除
        LEO  end    每個(gè)副本的【下一個(gè)待寫入】的offset值 
        HW  高水位線  只有它之前的消息才能被消費(fèi)者拉取消費(fèi)
            hw=min(leo)  類似于木桶效應(yīng)
  

Producer生產(chǎn)者如何保證生產(chǎn)的數(shù)據(jù)不丟失\不重復(fù)

• 如何保證kafka數(shù)據(jù)不丟失
  ack校檢和重試機(jī)制

  • ACK級(jí)別
    0 不管ack 只發(fā)送數(shù)據(jù)
    1 當(dāng)leader副本保存成功 返回ack給生產(chǎn)者
    -1|all 當(dāng)所有的ISR副本都保存同步成功 返回ack 【最安全 最慢】
  • 重試機(jī)制
    因?yàn)榫W(wǎng)絡(luò)質(zhì)量等偶發(fā)因素導(dǎo)致的消息發(fā)送失敗 可以通過重試機(jī)制
    如果因?yàn)榇a問題 集群環(huán)境問題 重試一百萬(wàn)次有沒有意義

• 如何保證數(shù)據(jù)寫入kafka不重復(fù)？
  為什么會(huì)重復(fù)？
  1_重試機(jī)制，如果前一個(gè)ack還沒有返回 生產(chǎn)者認(rèn)為失敗了基于重試機(jī)制重新發(fā)了一遍
  2_ack結(jié)果延遲 丟失
  實(shí)際上kafka已經(jīng)存儲(chǔ)成功了 只不過生產(chǎn)者沒有正確準(zhǔn)確的收到ack

  內(nèi)部設(shè)計(jì)了什么機(jī)制？
  1_冪等性機(jī)制 操作一次和操作多次 效果是一樣的。 不跟次數(shù)有關(guān)。 重復(fù)支付 重復(fù)提交 常量函數(shù)。
  2_給每個(gè)生產(chǎn)者發(fā)送的消息內(nèi)部編號(hào) 自增id
  kafka在保存數(shù)據(jù)的時(shí)候 就會(huì)判斷編號(hào) 如果已經(jīng)有了 不保存了直接返回ack

生產(chǎn)者寫入數(shù)據(jù)分區(qū)規(guī)則

當(dāng)Producer生產(chǎn)者向Topic隊(duì)列中發(fā)送數(shù)據(jù)時(shí)，如何確定發(fā)送到哪個(gè)分區(qū)Partition呢？

• 1.如果用戶指定了分區(qū),就向指定分寫入數(shù)據(jù)

• 2.如果用戶不指定分區(qū),看是否有自定義分區(qū)規(guī)則

  • 2_1 如果沒有指定自定義分區(qū)規(guī)則,按照默認(rèn)的規(guī)則分區(qū)規(guī)則
    如果有key,根據(jù) key的hash值%分區(qū)個(gè)數(shù)

  Utils.murmur2(keyBytes) % numPartitions
  計(jì)算key哈希值，對(duì)partition分區(qū)個(gè)數(shù)進(jìn)行取模操作 結(jié)果就是分區(qū)編號(hào)。
  只要key一樣，一定到同一個(gè)分區(qū)。

  如果沒有key:

  a_老版本 采用輪詢策略
  b_新版本 stickyPartition 黏性策略
  if (keyBytes == null) {
  return stickyPartitionCache.partition(topic, cluster);
  }

  什么叫做黏性策略呢？
  首先判斷當(dāng)前有沒有partition的連接 如果連接有效 直接使用這個(gè)連接
  Integer newPart = oldPart;
  如果沒有分區(qū)連接，那就隨機(jī)選擇一個(gè)分區(qū)創(chuàng)建連接 把數(shù)據(jù)都寫入這個(gè)分區(qū)
  random

  • 2_2 如果有自定義分區(qū)規(guī)則,按照自定義分區(qū)規(guī)則分配

Consumer消費(fèi)者如何保證數(shù)據(jù)不重復(fù)不丟失

Kafka記錄了每次消費(fèi)者消費(fèi)后的消費(fèi)記錄
當(dāng)消費(fèi)者來(lái)消費(fèi)的時(shí)候,只需消費(fèi)上一次offset+1的數(shù)據(jù)就可以了

• 消費(fèi)者消費(fèi)數(shù)據(jù)的三種方式
  指定topic ???? 讀取的是指定topic下所有分區(qū)的數(shù)據(jù)
  指定topic partition 讀取的是指定topic下的某個(gè)分區(qū)的數(shù)據(jù)
  指定topic partition offset 讀取的是指定topic下 指定分區(qū)的 某個(gè)offset開始

訂閱主題 subscribe 消費(fèi)該主題的所有分區(qū)數(shù)據(jù)
訂閱主題指定分區(qū) assign 消費(fèi)指定主題分區(qū)的數(shù)據(jù)
精準(zhǔn)消費(fèi) 定位消費(fèi) seek 消費(fèi)指定主題分區(qū) 指定偏移量消費(fèi)
需要先訂閱topic 和 分區(qū),然后才能seek 偏移量

消費(fèi)者開始消費(fèi)數(shù)據(jù)時(shí),從哪里開始消費(fèi)

• step1: 第一次消費(fèi)規(guī)則 由屬性決定
  lastest 從0開始
  earliest 從最新的開始(默認(rèn))

      //設(shè)置消費(fèi)的位置 從哪里開始消費(fèi)  合法參數(shù)：latest |  earliest
      props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

• step2:第二次及以后 從上一次消費(fèi)完的offset +1位置開始消費(fèi)

• 以上成為消費(fèi)記錄,那么kafka如何保存消費(fèi)記錄的呢?
  消費(fèi)者內(nèi)存中 自己維護(hù) 程序不斷 不重啟的情況下 【優(yōu)先使用】。
  持久化存儲(chǔ)在磁盤、存儲(chǔ)介質(zhì)： 【重啟程序的依據(jù)】
  存儲(chǔ)kafka中（默認(rèn)） 自創(chuàng)了主題_consumer_offsets
  存儲(chǔ)zk/mysql/redis

內(nèi)存消費(fèi)記錄:kafka自己記錄的非常標(biāo)準(zhǔn),但是程序重啟之后就沒有了,需要從文件當(dāng)中讀

文件消費(fèi)記錄:kafka自己有一個(gè)topic 專門用來(lái)存儲(chǔ)每個(gè)消費(fèi)者消費(fèi)的offset

image.png

offset偏移量管理
消費(fèi)者消費(fèi)完的記錄需要提交,怎樣提交?

• 自動(dòng)提交
  • 根據(jù)時(shí)間周期提交下一下的消費(fèi)的offset,默認(rèn)每五秒提交一次

  • 風(fēng)險(xiǎn):
    數(shù)據(jù)丟失
    數(shù)據(jù)重復(fù)

    
    
    image.png

為了防止數(shù)據(jù)丟失,或者重復(fù)消費(fèi)我們選擇手動(dòng)提交

• 手動(dòng)提交offset
  先消費(fèi)數(shù)據(jù),然后再提交offset
  風(fēng)險(xiǎn):如果此時(shí)我已經(jīng)消費(fèi)了兩個(gè)分區(qū)的數(shù)據(jù),第三個(gè)分區(qū)還沒有消費(fèi)完,程序崩潰了,offset沒有提交,就會(huì)導(dǎo)致,下次程序啟動(dòng)的時(shí)候 重復(fù)消費(fèi)數(shù)據(jù)
• 手動(dòng)提交--基于分區(qū)提交offset
  offsets.put(partition,new OffsetAndMetadata(consumerOffset+1));
  把當(dāng)前分區(qū)消費(fèi)的最后一條日志的offset +1,提交上去。

• 手動(dòng)提交分區(qū)offset探秘

  
  
  image.png

消費(fèi)者在自己的內(nèi)存中維護(hù)了消費(fèi)記錄
當(dāng)內(nèi)存中有記錄的時(shí)候,程序之間從內(nèi)存當(dāng)中讀取消費(fèi)記錄,這個(gè)消費(fèi)記錄是自己維護(hù)的正確的

而磁盤中的消費(fèi)記錄,只有在程序重啟 內(nèi)存中的消費(fèi)記錄丟失了 才會(huì)根據(jù)磁盤去消費(fèi)

消費(fèi)者消費(fèi)數(shù)據(jù)分配策略

問題：Kafka 消費(fèi)組Consumer Group中多個(gè)消費(fèi)者Consumer如何消費(fèi)Topic隊(duì)列中數(shù)據(jù)？
前提-kafka中同一個(gè)消費(fèi)組中的消費(fèi)者規(guī)則:
一個(gè)分區(qū)只能被一個(gè)消費(fèi)者消費(fèi)
一個(gè)消費(fèi)者可以消費(fèi)多個(gè)分區(qū)

另 一個(gè)分區(qū)的內(nèi)容,可以被不同消費(fèi)組內(nèi)的消費(fèi)者消費(fèi)
最理想的狀態(tài):消費(fèi)者和分區(qū) 一對(duì)一
策略:
范圍分配
輪詢分配
粘性分配

1_RangeAssignor 分配策略-范圍分配

• Kafka中默認(rèn)的分配規(guī)則
• 一個(gè)topic中所有的分區(qū)按照消費(fèi)者的個(gè)數(shù)平均分,多的就分配給編號(hào)小的消費(fèi)者
  優(yōu)點(diǎn):適用于消費(fèi)topic比較少的情況,分配會(huì)比較平均
  ex: 一個(gè)topic內(nèi)有七個(gè)分區(qū) 有三個(gè)消費(fèi)者c1 c2 c3
  c1 (0,1,2) c2(3,4) c3(56)
  缺點(diǎn): 不適應(yīng)于多個(gè)消費(fèi)者消費(fèi)多個(gè)topic,會(huì)造成編號(hào)小的負(fù)載壓力大的情況
  列如:三個(gè)消費(fèi)者 消費(fèi)三個(gè)topic 每個(gè)topic有7個(gè)分區(qū)會(huì)導(dǎo)致
  c1 -t1(0,1,2) t2(0,1,2) t3(0,1,2)

c2 -t1(3,4) t2(3,4) t3(3,4)
c3 t1(5,6) t2(5,6) t3(5,6)

2_RoundRobinAssignor 分配策略-輪詢策略

• 給每個(gè)Topic和其分區(qū)編號(hào),輪詢分配給消費(fèi)者
  一個(gè)消費(fèi)者分配一個(gè) 輪詢分配
  適合：所有消費(fèi)者都訂閱相同的主題。
  缺點(diǎn)：如果有消費(fèi)者故障 或者加入新的消費(fèi)者 之前全推倒 重新分配

image.png