目錄
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ù)積壓
通過命令的方式查看數(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ù)器無法存儲(chǔ),利用多臺(tái)服務(wù)器的存儲(chǔ)能力,提升Topic的數(shù)據(jù)存儲(chǔ)條數(shù)
2.提升Topic的吞吐量(數(shù)據(jù)讀寫速度):利用多臺(tái)服務(wù)器的能力,網(wǎng)絡(luò)等資源
分區(qū)數(shù)量
分區(qū)的數(shù)量沒有限制,分區(qū)數(shù)量和Kafka集群中的broker節(jié)點(diǎn)個(gè)數(shù)沒有任何關(guān)系,在實(shí)際工作推薦Topic的分區(qū)數(shù)量不要超過kafka集群中的broker節(jié)點(diǎn)個(gè)數(shù)的3倍,這只是一個(gè)推薦/經(jīng)驗(yàn)值
副本
副本作用:
通過多副本的機(jī)制,提升數(shù)據(jù)安全性,但是副本過多,會(huì)導(dǎo)致冗余(重復(fù))的數(shù)據(jù)過多
副本的數(shù)量限制
副本數(shù)量最大不能超過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文件:稱之為索引文件,該文件的作用是用來加快對(duì)xx.log文件內(nèi)容檢索的速度
xx.log和xx.index文件名稱的意義:
這個(gè)數(shù)字是xx.log文件中第一條消息的offset(偏移量),offset(偏移量)從0開始編號(hào)
partition分區(qū)的數(shù)據(jù)分成多個(gè)xx.log(segment片段文件)文件進(jìn)行存儲(chǔ)的意義:
1.如果一個(gè)文件數(shù)據(jù)量過大,打開和關(guān)閉文件都非常消耗資源
2.在一個(gè)大文件中,檢索內(nèi)容也會(huì)非常消耗資源
3.kafka只是用來臨時(shí)存儲(chǔ)消息數(shù)據(jù),會(huì)定時(shí)將過期數(shù)據(jù)刪除,如果數(shù)據(jù)在一個(gè)文件中,刪除效率低,如果數(shù)據(jù)分成了多個(gè)segment片段文件進(jìn)行存儲(chǔ),刪除的時(shí)候只需要判斷segment文件最后修改時(shí)間,如果超過了保留時(shí)間,就直接將整個(gè)segment文件刪除,該保留時(shí)間是通過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，沒有指定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)沒有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開始
當(dāng)指定了partition的參數(shù)后, 與DefaultPartitioner沒有任何的關(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ù)傾斜的問題。
在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但是沒有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但是沒有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ù)最多不能超過Topic的分區(qū)數(shù),如果超過了,就會(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)組間沒有任何關(guān)系,也就是Topic的數(shù)據(jù)能夠被對(duì)個(gè)消費(fèi)組中的消費(fèi)者重復(fù)消費(fèi)
查看消費(fèi)組中有多少個(gè)消費(fèi)者，用來避免消費(fèi)者個(gè)數(shù)超過分區(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ù)類型：string2- buffer.memory 緩存大小
??? 默認(rèn)值：33554432（32MB）3- retries 失敗后重試次數(shù)
??? 默認(rèn)值：2147483647，該值沒有意義，一般是使用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端通過多副本機(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.如果沒有查到offset,消費(fèi)者默認(rèn)從Topic最新的地方開始消費(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)的過程中,底層有個(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來保存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)者來確保消息只被發(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的值,一直在增大沒有在減小,正常情況下,lag的值是來回波動(dòng)的
通過命令的方式查看數(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)邏輯過于復(fù)雜,導(dǎo)致消費(fèi)過慢,出現(xiàn)積壓?jiǎn)栴}
解決方案:
1.處理異常容器,保證一直可用狀態(tài)
2.對(duì)于第二種, 如果之前一直沒問題, 只是某一天出現(xiàn), 可以調(diào)整消費(fèi)的超時(shí)時(shí)間。并且同時(shí)解決網(wǎng)絡(luò)延遲問題文章來源：http://www.zghlxwxcb.cn/news/detail-796121.html
3.對(duì)于第三種,調(diào)整消費(fèi)代碼,消費(fèi)更快,利用消費(fèi)者的負(fù)載均衡策略,提升消費(fèi)者數(shù)量文章來源地址http://www.zghlxwxcb.cn/news/detail-796121.html
到了這里，關(guān)于Kafka的核心原理的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
