柚子快報(bào)激活碼778899分享：分布式 Kafka的核心原理

http://yzkb.51969.com/

目錄

Tpoic的分區(qū)和副本機(jī)制

分區(qū)

副本

?消息存儲(chǔ)機(jī)制和查詢機(jī)制

消息存儲(chǔ)機(jī)制

? log文件和index文件的解析

?index文件內(nèi)容基本結(jié)構(gòu)

查詢機(jī)制

Kafka中生產(chǎn)者數(shù)據(jù)分發(fā)策略

隨機(jī)分發(fā)策略

指定分區(qū)策略

Hash取模策略

自定義分區(qū)策略

輪詢分發(fā)策略 和 粘性分發(fā)策略

Kafka消費(fèi)者的負(fù)載均衡機(jī)制

數(shù)據(jù)不丟失機(jī)制

生產(chǎn)端保證消息不丟失

Broker端保證數(shù)據(jù)不丟失

消費(fèi)端如何保證數(shù)據(jù)不丟失

Kafka的數(shù)據(jù)積壓

通過(guò)命令的方式查看數(shù)據(jù)積壓

?數(shù)據(jù)積壓?jiǎn)栴}處理

Tpoic的分區(qū)和副本機(jī)制

分區(qū)

分區(qū)的作用:

1.避免單臺(tái)服務(wù)器容量的限制:每臺(tái)服務(wù)器的磁盤存儲(chǔ)空間是有限的,Topic分成多個(gè)partition分區(qū),可以避免單個(gè)partition數(shù)據(jù)大小多大,導(dǎo)致服務(wù)器無(wú)法存儲(chǔ),利用多臺(tái)服務(wù)器的存儲(chǔ)能力,提升Topic的數(shù)據(jù)存儲(chǔ)條數(shù)

2.提升Topic的吞吐量(數(shù)據(jù)讀寫速度):利用多臺(tái)服務(wù)器的能力,網(wǎng)絡(luò)等資源

分區(qū)數(shù)量

分區(qū)的數(shù)量沒(méi)有限制,分區(qū)數(shù)量和Kafka集群中的broker節(jié)點(diǎn)個(gè)數(shù)沒(méi)有任何關(guān)系,在實(shí)際工作推薦Topic的分區(qū)數(shù)量不要超過(guò)kafka集群中的broker節(jié)點(diǎn)個(gè)數(shù)的3倍,這只是一個(gè)推薦/經(jīng)驗(yàn)值

副本

副本作用:

通過(guò)多副本的機(jī)制,提升數(shù)據(jù)安全性,但是副本過(guò)多,會(huì)導(dǎo)致冗余(重復(fù))的數(shù)據(jù)過(guò)多

副本的數(shù)量限制

副本數(shù)量最大不能超過(guò)kafka集群中的broker節(jié)點(diǎn)個(gè)數(shù),在實(shí)際工作中,推薦的分區(qū)的副本數(shù)量是1-3個(gè),具體設(shè)置多少個(gè),根據(jù)企業(yè)的數(shù)據(jù)重要程度進(jìn)行選擇,如果數(shù)據(jù)重要,可以將副本數(shù)設(shè)置大一些,如果數(shù)據(jù)不太重要,可以將副本數(shù)設(shè)置小一些

?消息存儲(chǔ)機(jī)制和查詢機(jī)制

消息存儲(chǔ)機(jī)制

?1.Topic的數(shù)據(jù)存放路徑是:/export/server/kafka/data,在該目錄下,還有其他的目錄,而且以Topic進(jìn)行劃分,具體目錄的命名規(guī)則是:Topic名稱-分區(qū)編號(hào)

2.Topic目錄下,存放的是消息的數(shù)據(jù)文件,并且是成對(duì)出現(xiàn),也就是xx.log文件和xx.index文件

? log文件和index文件的解析

xx.log文件和xx.index文件的作用:

xx.log文件:稱之為segment片段文件,也就是一個(gè)partition分區(qū)的數(shù)據(jù),會(huì)被分成多個(gè)segment(log)片段文件進(jìn)行存儲(chǔ)

xx.index文件:稱之為索引文件,該文件的作用是用來(lái)加快對(duì)xx.log文件內(nèi)容檢索的速度

xx.log和xx.index文件名稱的意義:

這個(gè)數(shù)字是xx.log文件中第一條消息的offset(偏移量),offset(偏移量)從0開(kāi)始編號(hào)

partition分區(qū)的數(shù)據(jù)分成多個(gè)xx.log(segment片段文件)文件進(jìn)行存儲(chǔ)的意義:

1.如果一個(gè)文件數(shù)據(jù)量過(guò)大,打開(kāi)和關(guān)閉文件都非常消耗資源

2.在一個(gè)大文件中,檢索內(nèi)容也會(huì)非常消耗資源

3.kafka只是用來(lái)臨時(shí)存儲(chǔ)消息數(shù)據(jù),會(huì)定時(shí)將過(guò)期數(shù)據(jù)刪除,如果數(shù)據(jù)在一個(gè)文件中,刪除效率低,如果數(shù)據(jù)分成了多個(gè)segment片段文件進(jìn)行存儲(chǔ),刪除的時(shí)候只需要判斷segment文件最后修改時(shí)間,如果超過(guò)了保留時(shí)間,就直接將整個(gè)segment文件刪除,該保留時(shí)間是通過(guò)server.properties文件中的log.retention.hours=168進(jìn)行設(shè)置的,默認(rèn)保留168小時(shí)(7天)

?index文件內(nèi)容基本結(jié)構(gòu)

查詢機(jī)制

查詢步驟:,

1.首先先確定要讀取哪個(gè)xx.log(segment片段)文件,368776該offset的消息在368769.log文件中

2.查詢xx.log對(duì)應(yīng)的xx.index,查詢?cè)摋l消息的物理偏移量范圍

3.根據(jù)消息的物理偏移量范圍去讀取xx.log文件(底層是基于磁盤的順序讀取)

4.最終就獲取到了具體的消息內(nèi)容

Kafka中生產(chǎn)者數(shù)據(jù)分發(fā)策略

? 生產(chǎn)者數(shù)據(jù)分發(fā)策略指的集市生產(chǎn)者生產(chǎn)的消息是如何保存到具體分區(qū)上

分發(fā)策略如下分類:

1.隨機(jī)分發(fā)策略:將消息發(fā)到隨機(jī)某個(gè)分區(qū)上,還是發(fā)送到Leader主副本上,python支持,java不支持

2.指定分區(qū)策略:將消息發(fā)到指定分區(qū)上面python支持,java不支持

3.Hash取模策略:對(duì)消息的key先取Hash值,再和分區(qū)數(shù)取模,python支持,java不支持

4.輪詢策略:再kafka的2.4及以上版本,已經(jīng)更名成粘性分發(fā)策略,python不支持,java支持

5.自定義分發(fā)策略:,python支持,java支持

隨機(jī)分發(fā)策略

def send(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):

當(dāng)在發(fā)送數(shù)據(jù)的時(shí)候, 如果只傳遞了topic 和 value，沒(méi)有指定key的時(shí)候, 那么此時(shí)就采用隨機(jī)策略

在kafka中, 專門提供了一個(gè)默認(rèn)的分發(fā)數(shù)據(jù)的類: DefaultPartitioner ??? def __call__(cls, key, all_partitions, available): ??????? """ ?????? ??? ?如果 key為 null, 那么隨機(jī)返回一個(gè)分區(qū)的編號(hào) ??????? """ ??????? if key is None: ??????????? if available: ??????????????? return random.choice(available) ??????????? return random.choice(all_partitions) ?? ??? ?# 后續(xù)的代碼 當(dāng)沒(méi)有key的時(shí)候,壓根就執(zhí)行不到 ??????? idx = murmur2(key) ??????? idx &= 0x7fffffff ??????? idx %= len(all_partitions) ??????? return all_partitions[idx]

指定分區(qū)策略

def send(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):

當(dāng)在發(fā)送數(shù)據(jù)的時(shí)候, 如果指定了partition參數(shù), 表示的采用指定分區(qū)的方案, 分區(qū)的編號(hào)從0開(kāi)始

當(dāng)指定了partition的參數(shù)后, 與DefaultPartitioner沒(méi)有任何的關(guān)系

Hash取模策略

def send(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):

當(dāng)在發(fā)送數(shù)據(jù)的時(shí)候, 如果傳遞了topic 和 value 以及key的時(shí)候, 那么此時(shí)就是采用hash取模策略

注意: 相同key的返回的hash值是一致的, 同樣對(duì)應(yīng)分區(qū)也是同一個(gè)。也就是要注意數(shù)據(jù)傾斜的問(wèn)題。

在kafka中, 專門提供了一個(gè)默認(rèn)的分發(fā)數(shù)據(jù)的類: DefaultPartitioner ??? def __call__(cls, key, all_partitions, available): ??????? """ ?????? ??? ?如果 key為 null, 那么隨機(jī)返回一個(gè)分片的編號(hào) ??????? """ ??????? if key is None: ??????????? if available: ??????????????? return random.choice(available) ??????????? return random.choice(all_partitions) ?? ??? ?# 當(dāng)有key的時(shí)候 , 執(zhí)行下列代碼. 此處的代碼其實(shí)就是hash取模的方案 ??????? idx = murmur2(key) ??????? idx &= 0x7fffffff ??????? idx %= len(all_partitions) ??????? return all_partitions[idx]

自定義分區(qū)策略

參考DefaultPartitioner 它怎么寫你就怎么寫

# 第一步：創(chuàng)建自己的分區(qū)類 class MyPartitioner(object):

?? ?# 第二步：實(shí)現(xiàn)__call__。key：消息中的key；all_partitions：所有的分區(qū)列表；available：所有可用分區(qū)的列表 ??? @classmethod ??? def __call__(cls, key, all_partitions, available): ??????? # 第三步：分發(fā)邏輯根據(jù)自己要求進(jìn)行實(shí)現(xiàn) ?????? return 0 ????? ? # 第四步：導(dǎo)入自己的分區(qū)類 import MyPartitioner

# 第五步：調(diào)用 producer = KafkaProducer( ?? ?bootstrap_servers=['node1.itcast.cn:9092','node2.itcast.cn:9092'], ?? ?partitioner=MyPartitioner() )

輪詢分發(fā)策略 和 粘性分發(fā)策略

輪詢分發(fā)策略: 在Kafka的老版本中存在的一種分發(fā)策略，當(dāng)生產(chǎn)數(shù)據(jù)的時(shí)候，只有value但是沒(méi)有key的時(shí)候，采用輪詢。 ?? ?優(yōu)點(diǎn): 可以保證每個(gè)分區(qū)拿到的數(shù)據(jù)基本是一樣，因?yàn)槭且粋€(gè)一個(gè)的輪詢的分發(fā) ?? ?缺點(diǎn): 如果采用異步發(fā)送方式，意味著一批數(shù)據(jù)發(fā)送到broker端，由于是輪詢策略，會(huì)將這一批數(shù)據(jù)拆分為多個(gè)小的批次，分別再寫入到不同的分區(qū)里面去，寫入進(jìn)去以后，每個(gè)分區(qū)都會(huì)給予響應(yīng)，會(huì)影響寫入效率。 ?? ? 粘性分發(fā)策略: 在Kafka新版本中存在的一種分發(fā)策略。當(dāng)生產(chǎn)數(shù)據(jù)的時(shí)候，只有value但是沒(méi)有key的時(shí)候，采用粘性分發(fā)策略 ?? ?優(yōu)點(diǎn): 在發(fā)送數(shù)據(jù)的時(shí)候，首先會(huì)隨機(jī)的選取一個(gè)分區(qū)，然后盡可能將數(shù)據(jù)分發(fā)到這個(gè)分區(qū)上面去，也就是盡可能粘著這個(gè)分區(qū)。該分發(fā)方式，在異步發(fā)送的操作中，效率比較高。 ?? ?缺點(diǎn): 在數(shù)據(jù)發(fā)送特別快的時(shí)候，可能會(huì)導(dǎo)致某個(gè)分區(qū)的數(shù)據(jù)比其他分區(qū)數(shù)據(jù)多很多，造成大量的數(shù)據(jù)集中在一個(gè)分區(qū)上面

Kafka消費(fèi)者的負(fù)載均衡機(jī)制

1.在同一個(gè)消費(fèi)組中,消費(fèi)者的個(gè)數(shù)最多不能超過(guò)Topic的分區(qū)數(shù),如果超過(guò)了,就會(huì)有一些消費(fèi)者處于閑置狀態(tài),消費(fèi)不到任何數(shù)據(jù)

2.在同一個(gè)消費(fèi)組中,一個(gè)Topic中一個(gè)分區(qū)的數(shù)據(jù),只能被同個(gè)消費(fèi)組中的一個(gè)消費(fèi)者所消費(fèi),不能被同個(gè)消費(fèi)者組中多個(gè)消費(fèi)者所消費(fèi),但是一個(gè)消費(fèi)組可以消費(fèi)多個(gè)分區(qū)的數(shù)據(jù),也就是分區(qū)和消費(fèi)的對(duì)應(yīng)關(guān)系,多對(duì)一

3.不同的消費(fèi)組中的消費(fèi)者,可以對(duì)一個(gè)Topic的數(shù)據(jù)同時(shí)消費(fèi),也就是不同消費(fèi)組間沒(méi)有任何關(guān)系,也就是Topic的數(shù)據(jù)能夠被對(duì)個(gè)消費(fèi)組中的消費(fèi)者重復(fù)消費(fèi)

查看消費(fèi)組中有多少個(gè)消費(fèi)者，用來(lái)避免消費(fèi)者個(gè)數(shù)超過(guò)分區(qū)個(gè)數(shù)。

./kafka-consumer-groups.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --group g_1 --members --describe

數(shù)據(jù)不丟失機(jī)制

生產(chǎn)端保證消息不丟失

生產(chǎn)者端保證數(shù)據(jù)不丟失:

? 生產(chǎn)者端將消失發(fā)送給到kafka集群以后,broker要給生產(chǎn)者響應(yīng)信息,響應(yīng)原理就是ACK機(jī)制

ACK機(jī)制當(dāng)中有3個(gè)參數(shù)配置值,分別是: 0? 1? -1 (all)

0 :生產(chǎn)者生產(chǎn)消息給到kafka集群,生產(chǎn)者不等待(不接收)broker返回的響應(yīng)信息

1 :生產(chǎn)者生產(chǎn)消息給到kafka集群,kafka集群中的分區(qū)對(duì)應(yīng)的Leader主副本所在的broker給生產(chǎn)者返回響應(yīng)信息

-1(all) :生產(chǎn)者生產(chǎn)消息給到kafka集群,kafka集群中的分區(qū)對(duì)應(yīng)的所有副本給生產(chǎn)者返回響應(yīng)信息

消息的生產(chǎn)效率排序(由高到低):0>1>-1

消息的安全級(jí)別排序(由高到低):-1>1>0

在實(shí)際工作中如何選擇ACK參數(shù)配置？ 答：根據(jù)數(shù)據(jù)的重要程度進(jìn)行選擇。如果數(shù)據(jù)重要，優(yōu)先保證數(shù)據(jù)的安全性，再考慮生產(chǎn)效率；如果數(shù)據(jù)不重要，優(yōu)先考慮生產(chǎn)效率，再盡可能提升安全級(jí)別。

相關(guān)的參數(shù)

1- acks? broker節(jié)點(diǎn)確認(rèn)機(jī)制 ??? 默認(rèn)值：1；數(shù)據(jù)類型：string

2- buffer.memory 緩存大小 ??? 默認(rèn)值：33554432（32MB）

3- retries 失敗后重試次數(shù) ??? 默認(rèn)值：2147483647，該值沒(méi)有意義，一般是使用delivery.timeout.ms參數(shù)進(jìn)行控制

4- delivery.timeout.ms 消息傳輸超時(shí)時(shí)間 ??? 默認(rèn)值：120000（120秒）

5- batch.size 每一批次的消息數(shù)據(jù)的大小 ??? 默認(rèn)值：16384（16KB）

6- linger.ms 每一批次的間隔時(shí)間 ??? 默認(rèn)值：0

Broker端保證數(shù)據(jù)不丟失

? Broker端通過(guò)多副本機(jī)制確保數(shù)據(jù)不丟失。同時(shí)需要生產(chǎn)者端將acks設(shè)置為-1

消費(fèi)端如何保證數(shù)據(jù)不丟失

消費(fèi)者消費(fèi)消息的步驟:

1.消費(fèi)者首先連接到kafka集群中,進(jìn)行消息的消費(fèi)

2.kafka集群接收到Consumer消費(fèi)者的消費(fèi)請(qǐng)求后,首先會(huì)根據(jù)group id(消費(fèi)組名稱),查找上次消費(fèi)消息對(duì)應(yīng)的offset(偏移量)

3.如果沒(méi)有查到offset,消費(fèi)者默認(rèn)從Topic最新的地方開(kāi)始消費(fèi)

4.如果有查到offset,會(huì)從上次消費(fèi)到的offset地方進(jìn)行繼續(xù)消費(fèi)

??????? 4-1.首先先確定要讀取的這個(gè)offset偏移量在哪個(gè)segment文件中

??????? 4-2.查詢這個(gè)segment文件對(duì)應(yīng)的index文件,根據(jù)offset確定這個(gè)消息在log文件的什么位置,也就是確定消息的物理偏移量

??????? 4-3:讀取log文件,查詢對(duì)應(yīng)范圍內(nèi)的數(shù)據(jù)即可

??????? 4-4:獲取最終的消息數(shù)據(jù)

5.消費(fèi)者在消費(fèi)的過(guò)程中,底層有個(gè)線程會(huì)定時(shí)的將消費(fèi)的offset提交給到kafka集群,kafka集群會(huì)更新對(duì)應(yīng)的offset的值

該流程能夠保證消費(fèi)端不丟失數(shù)據(jù)嗎？ ?????????可以保證消費(fèi)端數(shù)據(jù)不丟失。但是會(huì)出現(xiàn)重復(fù)消費(fèi)的情況。

消費(fèi)組的offset信息保存在什么地方？ ????????Kafka集群內(nèi)部會(huì)創(chuàng)建一個(gè)叫做__consumer_offsets的Topic來(lái)保存offset信息。該Topic有50個(gè)分區(qū)，1個(gè)副本

Kafka中消費(fèi)者如何對(duì)數(shù)據(jù)僅且只消費(fèi)一次？

1- 將消費(fèi)者的 enable.auto.commit 屬性設(shè)置為 false，并手動(dòng)管理消費(fèi)者的偏移量。這樣可以確保消費(fèi)者在處理完所有消息后才更新偏移量，避免重復(fù)消費(fèi)數(shù)據(jù)。也就是將消息的消費(fèi)、消息業(yè)務(wù)處理代碼、offset提交代碼放在同一個(gè)事務(wù)當(dāng)中。

2- 使用冪等生產(chǎn)者或事務(wù)性生產(chǎn)者來(lái)確保消息只被發(fā)送一次。這樣可以避免重復(fù)發(fā)送消息，從而避免消費(fèi)者重復(fù)消費(fèi)數(shù)據(jù)。

3- 在消息中加入唯一的ID

在提交偏移量的時(shí)候，有二種提交方式: 自動(dòng)提交偏移量 和 手動(dòng)提交偏移量，手動(dòng)提交又分了同步和異步

Kafka的數(shù)據(jù)積壓

? 數(shù)據(jù)持續(xù)在kafka集群中積壓,也就是lag的值,一直在增大沒(méi)有在減小,正常情況下,lag的值是來(lái)回波動(dòng)的

通過(guò)命令的方式查看數(shù)據(jù)積壓

kafka集群有哪些消費(fèi)組

./kafka-consumer-groups.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --list

查看指定消費(fèi)組數(shù)據(jù)積壓情況

./kafka-consumer-groups.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092? --describe --group g_2

?數(shù)據(jù)積壓?jiǎn)栴}處理

出現(xiàn)積壓的原因:

1.因?yàn)閷懭肽康娜萜魇?從而導(dǎo)致消費(fèi)失敗

2.因?yàn)榫W(wǎng)絡(luò)延遲消費(fèi)失敗

3.消費(fèi)邏輯過(guò)于復(fù)雜,導(dǎo)致消費(fèi)過(guò)慢,出現(xiàn)積壓?jiǎn)栴}

解決方案:

1.處理異常容器,保證一直可用狀態(tài)

2.對(duì)于第二種, 如果之前一直沒(méi)問(wèn)題, 只是某一天出現(xiàn), 可以調(diào)整消費(fèi)的超時(shí)時(shí)間。并且同時(shí)解決網(wǎng)絡(luò)延遲問(wèn)題

3.對(duì)于第三種,調(diào)整消費(fèi)代碼,消費(fèi)更快,利用消費(fèi)者的負(fù)載均衡策略,提升消費(fèi)者數(shù)量

柚子快報(bào)激活碼778899分享：分布式 Kafka的核心原理

http://yzkb.51969.com/

精彩文章