github地址：https://github.com/douzixiansheng/MQ/blob/master/kafka_basic.md
Kafka
Kafka是一個(gè)分布式消息隊(duì)列厌秒，具有高性能夹姥、持久化、多副本備份顽耳、橫向擴(kuò)展能力.(pub-sub模型)
維基百科
- Kafka 是由Apache軟件基金會(huì)開發(fā)的一個(gè)開源流處理平臺(tái)恰梢，由Scala和JAVA編寫.該項(xiàng)目的目標(biāo)是為處理實(shí)時(shí)數(shù)據(jù)提供一個(gè)統(tǒng)一近刘、高吞吐搔谴、低延遲的平臺(tái)喉祭。其持久化層本質(zhì)是一個(gè)"按照分布式事務(wù)日志架構(gòu)的大規(guī)模發(fā)布/訂閱消息隊(duì)列".Kafka可以通過Kafka Connect連接到外部系統(tǒng)(用于數(shù)據(jù)輸入/輸出),并提供了Kafka Streams ———— 一個(gè)Java流式處理庫(kù).
基于kafka-zookeeper 的分布式消息隊(duì)列系統(tǒng)總體架構(gòu)如下：
- kafka-zookeeper.png
Kafka 架構(gòu)說明
- 一個(gè)典型的Kafka集群包含若干Producer盖喷，若干Broker铃肯，若干Consumer，以及一個(gè)Zookeeper集群传蹈。Kafka通過Zookeeper管理集群配置押逼，選舉Leader，以及在Consumer Group發(fā)送變化時(shí)進(jìn)行Rebalance(負(fù)載均衡)惦界。Producer 使用push(推)模式將消息發(fā)布到Broker挑格；Consumer 使用pull(拉)模式從Broker訂閱并消費(fèi)消息。
Kafka 四大核心
- 生產(chǎn)者API：允許應(yīng)用程序發(fā)布記錄流至一個(gè)或多個(gè)kafka的主題(Topics)
- 消費(fèi)者API：允許應(yīng)用程序訂閱一個(gè)或多個(gè)主題沾歪，并處理這些主題接收到的記錄流
- Streams API: 允許應(yīng)用程序充當(dāng)流處理器(stream processor)漂彤，從一個(gè)或多個(gè)主題獲取輸入流，并生產(chǎn)一個(gè)輸出流至一個(gè)或多個(gè)的主題灾搏，能夠有效地變換輸入流為輸出流
- Connector API: 允許構(gòu)建和運(yùn)行可重用的生產(chǎn)者或消費(fèi)者挫望，能夠把kafka主題連接到現(xiàn)有的應(yīng)用程序或數(shù)據(jù)系統(tǒng)
Kafka 基礎(chǔ)概念
無論是kafka集群，還是consumer都依賴于zookeeper集群保存一些meta信息狂窑，來保證系統(tǒng)可用性
Producer
- 發(fā)送消息者稱為 Producer
Producer 采用推(push)模式將消息發(fā)布到broker媳板，每條消息都被追加(append)到分區(qū)(partition)中，屬于順序?qū)懘疟P(順序?qū)懘疟P效率比隨機(jī)寫內(nèi)存要高泉哈，保障kafka吞吐率)
生產(chǎn)者組件圖
- producer_base.png
- 創(chuàng)建Kafka生產(chǎn)者
- 要往Kafka寫入信息蛉幸，首先要?jiǎng)?chuàng)建一個(gè)生產(chǎn)者對(duì)象破讨，并設(shè)置一些屬性。Kafka生產(chǎn)者有3個(gè)必選屬性
| 參數(shù) | 描述 |
|---|---|
| bootstrap.servers | 該屬性指定broker的地址清單奕纫，地址的格式為host:port提陶。清單里不需要包含所有的broker地址，生產(chǎn)者會(huì)從給定的broker里查找到其他的broker的信息匹层。(建議提供兩個(gè)broker信息隙笆，一旦其中一個(gè)宕機(jī)，生產(chǎn)者仍然能夠連接到集群上) |
| key.serializer | broker 希望接受到的消息的鍵和值都是字節(jié)數(shù)組.生產(chǎn)者接口允許使用參數(shù)化類型升筏，因此可以把java對(duì)象作為鍵和值發(fā)送給broker仲器。 |
| value.serializer | value.serializer 指定的類會(huì)將值序列化。如果鍵和值都是字符串仰冠，可以使用key.serializer一樣的序列化器乏冀。如果鍵是整數(shù)類型而值是字符串，那么需要使用不同的序列化器. |
生產(chǎn)者的配置
| 參數(shù) | 描述 |
|---|---|
| acks | acks 參數(shù)指定了必須要有多少個(gè)分區(qū)副本收到消息洋只，生產(chǎn)者才會(huì)認(rèn)為消費(fèi)寫入是成功的辆沦。acks = 0 生產(chǎn)者在成功寫入消息之前不會(huì)等待任何來自服務(wù)器的響應(yīng)。(缺點(diǎn)：無法確認(rèn)消費(fèi)是否成功识虚；優(yōu)點(diǎn)：高吞吐量)肢扯；acks = 1 只要集群的首領(lǐng)節(jié)點(diǎn)收到消息，生產(chǎn)者就會(huì)收到一個(gè)來自服務(wù)器的成功響應(yīng)担锤。如果消費(fèi)無法到達(dá)首領(lǐng)節(jié)點(diǎn)(比如首領(lǐng)節(jié)點(diǎn)奔潰蔚晨，新的首領(lǐng)還沒有被選舉處理)，生產(chǎn)者會(huì)收到一個(gè)錯(cuò)誤響應(yīng)肛循，為了避免數(shù)據(jù)丟失铭腕，生產(chǎn)者會(huì)重發(fā)消息。不過多糠，如果一個(gè)沒有收到消息的節(jié)點(diǎn)成為新首領(lǐng)累舷，消息還是會(huì)丟失。acks = all 只有當(dāng)所有參與復(fù)制的節(jié)點(diǎn)全部收到消息時(shí)夹孔，生產(chǎn)者才會(huì)收到一個(gè)來自服務(wù)器的成功響應(yīng)被盈。 |
| buffer.memory | 該參數(shù)用來設(shè)置生產(chǎn)者緩沖區(qū)的大小，生產(chǎn)者用它緩沖要發(fā)送到服務(wù)器的消息搭伤。0.9.0.0 版本被替換成了 max.block.ms,表示在拋出異常之前可以阻塞一段時(shí)間 |
| compression.type | 默認(rèn)情況下為none只怎，消費(fèi)發(fā)送時(shí)不會(huì)被壓縮。該參數(shù)可以設(shè)置為snappy怜俐、gzip或lz4身堡，它指定了消息被發(fā)送給broker之前使用哪一種壓縮算法進(jìn)行壓縮。1. snappy 壓縮算法有Google發(fā)明佑菩，它占用較少的CPU盾沫，卻能提供較好的性能和相當(dāng)可觀的壓縮比(比較關(guān)注性能和網(wǎng)路帶寬) 2. gzip 壓縮算法一般會(huì)占用較多的CPU裁赠，但會(huì)提供更高的壓縮比(網(wǎng)絡(luò)帶寬有限次采用) |
| retries | 生產(chǎn)者從服務(wù)器收到的錯(cuò)誤有可能是臨時(shí)性的錯(cuò)誤(比如分區(qū)找不到首領(lǐng))殿漠。在這中情況下赴精，retries參數(shù)是值決定了生產(chǎn)者可以重發(fā)消息的次數(shù)，如果達(dá)到這個(gè)次數(shù)绞幌，生產(chǎn)者會(huì)放棄重試并返回錯(cuò)誤蕾哟。默認(rèn)情況下，生產(chǎn)者會(huì)在每次重試之間等待100ms莲蜘，可以通過retry.backoff.ms參數(shù)來改變這個(gè)時(shí)間間隔. |
| batch.size | 當(dāng)有多個(gè)消息需要被發(fā)送到同一個(gè)分區(qū)時(shí)谭确，生產(chǎn)者會(huì)把它們放在同一個(gè)批次里。該采納數(shù)指定了一個(gè)批次可以使用的內(nèi)存大小票渠，按照字節(jié)數(shù)計(jì)算(而不是消息個(gè)數(shù))逐哈。1. 批次設(shè)置很大 不會(huì)造成延遲，只會(huì)占用更多的內(nèi)存 2. 批次設(shè)置很小 因?yàn)樯a(chǎn)者需要更頻繁地發(fā)送消息问顷，會(huì)增加一些額外的開銷 |
| linger.ms | 該參數(shù)指定了生產(chǎn)者在發(fā)送批次之前等待更多消息加入批次的時(shí)間昂秃。 |
| client.id | 該參數(shù)可以是任意的字符串，服務(wù)器會(huì)用它來識(shí)別消息的來源杜窄，還可以用在日志和配額指標(biāo)里 |
| request.timeout.ms | 指定了生產(chǎn)者在發(fā)送數(shù)據(jù)時(shí)等待服務(wù)器返回響應(yīng)的時(shí)間 |
| max.block.ms | 該參數(shù)指定了在調(diào)用send()方法或使用partitionsFor()方法獲取元數(shù)據(jù)時(shí)生產(chǎn)者的阻塞時(shí)間肠骆。當(dāng)生產(chǎn)者的發(fā)送緩沖區(qū)已滿，或者沒有可用的元數(shù)據(jù)時(shí)塞耕，這些方法就會(huì)阻塞蚀腿。在阻塞時(shí)間達(dá)到max.block.ms時(shí)，生產(chǎn)者會(huì)拋出超時(shí)異常 |
| max.request.size | 該參數(shù)用于控制生產(chǎn)者發(fā)送的請(qǐng)求大小扫外。它可以指能發(fā)送的單個(gè)消息的最大值莉钙，可以指單個(gè)請(qǐng)求里所有消息總的大小。 |
- 生產(chǎn)者寫入流程
- producer_pull.png
Consumer
- 消息接收者稱為Consumer
- consumer 采用pull(拉)模式從broker中讀取數(shù)據(jù)
- push(推) 模式很難適應(yīng)消費(fèi)速率不同的消費(fèi)者筛谚，因?yàn)橄l(fā)送速率是由broker決定的胆胰。它的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成consumer來不及處理消息刻获，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞蜀涨。而pull模式則可以根據(jù)consumer的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息
- 對(duì)于Kafka而言,pull模式更合適，它可簡(jiǎn)化broker的設(shè)計(jì)蝎毡，consumer可自主控制消費(fèi)消息的速率厚柳，同時(shí)consumer可以自己控制消費(fèi)方式——即可批量消費(fèi)也可逐條消費(fèi)， 同時(shí)還能選擇不同的提交方式從而實(shí)現(xiàn)不同的傳輸語(yǔ)義
- pull 模式不足之處是沐兵，如果kafka沒有數(shù)據(jù)别垮，消費(fèi)者可能會(huì)陷入循環(huán)中，一直等待數(shù)據(jù)到達(dá)扎谎。為了避免這種情況碳想，我們?cè)诶?qǐng)求中有參數(shù)烧董，允許消費(fèi)者請(qǐng)求在的等待數(shù)據(jù)到達(dá)的"長(zhǎng)輪詢"中進(jìn)行阻塞(并且可選地等待到給定的字節(jié)數(shù)，以確保打的傳輸大小)
- Consumer Group (CG)
- 這是kafka用來實(shí)現(xiàn)一個(gè)topic消息的廣播(發(fā)給所有的consumer)和單播(發(fā)給任意一個(gè)consumer)的手段胧奔。一個(gè)topic可以有多個(gè)CG逊移。topic的消息會(huì)復(fù)制(概念上的復(fù)制)到所有的CG，但每個(gè)partition只會(huì)把消息發(fā)給該CG中的一個(gè)consumer龙填。如果需要實(shí)現(xiàn)廣播胳泉，只要每個(gè)consumer有一個(gè)獨(dú)立的CG就可以了。要實(shí)現(xiàn)單播只要所有consumer在同一個(gè)CG岩遗。用CG還可以將consumer進(jìn)行自由的分組而不需要多次發(fā)送消息到不同的topic
- 每個(gè)分區(qū)在同一時(shí)間只能由group中的一個(gè)消費(fèi)者讀取扇商，但是多個(gè)group可以同時(shí)消費(fèi)這個(gè)partition。
Broker(代理)
- 已發(fā)布的消息保存在一組服務(wù)器中宿礁，稱之為Kafka集群案铺。集群中的每一個(gè)服務(wù)器都是一個(gè)代理。
主題(Topic)
- Kafka將消息以topic為單位進(jìn)行歸納(一條消息必須屬于某一個(gè)主題)
- 在Kafka集群中梆靖，可以有無數(shù)的主題
- Kafka 的主題始終是支持多用戶訂閱的控汉；也就是說，一個(gè)主題可以有零個(gè)涤姊，一個(gè)或多個(gè)消費(fèi)者訂閱寫入的
數(shù)據(jù)
- Kafka 的主題始終是支持多用戶訂閱的控汉；也就是說，一個(gè)主題可以有零個(gè)涤姊，一個(gè)或多個(gè)消費(fèi)者訂閱寫入的
- 分區(qū)數(shù)(Partitions): 控制topic將分片成多少log赂鲤±甙危可以顯示指定戏自，如果不指定則會(huì)使用broker(server.properties)中的num.partitions配置的數(shù)量
- replication-factor副本：控制消息保證在幾個(gè)broker(服務(wù)器)上望浩，一般情況下等于broker的個(gè)數(shù)。
分區(qū)(Partitions)
- 消息發(fā)送時(shí)都被發(fā)送到一個(gè)topic恨课，其本質(zhì)就是一個(gè)目錄舆乔，而topic是由一些Partition Logs(分區(qū)日志)組成
- partitions_offset.png
- 每個(gè)Topic都有一個(gè)或者多個(gè)Partitions 構(gòu)成
- 每個(gè)Partition都是有序且不可變的消息隊(duì)列
- Topic的Partition數(shù)量可以在創(chuàng)建時(shí)配置
- Partition數(shù)量決定了每個(gè)Consumer group中并發(fā)消費(fèi)者的最大數(shù)量
- 分區(qū)的原因：
- 方便在集群中擴(kuò)展，每個(gè)Partition可以通過調(diào)整以適應(yīng)它所在的機(jī)器剂公，而一個(gè)topic又可以有多個(gè)Partition組成希俩，因此整個(gè)集群就可以適應(yīng)任意大小的數(shù)據(jù)了；
- 可以提高并發(fā)纲辽，因?yàn)榭梢砸訮artition為單位讀寫
- 分區(qū)的原則：
- 指定了partition颜武，則直接使用
- 未指定partition但指定key，通過對(duì)key的value進(jìn)行hash出一個(gè)partition
- partition和key都未指定拖吼，使用輪詢選出一個(gè)partition
偏移量(offset)
- 任何發(fā)布到partition的消息都會(huì)被直接追加到log文件的尾部鳞上，每條消息在文件中的位置稱為offset(偏移量),offset是一個(gè)long型數(shù)字，它唯一標(biāo)記一條消息吊档。消費(fèi)者通過(offset篙议、partition、topic)跟蹤記錄.
副本
- 副本因子操作的單位是以分區(qū)為單位,每個(gè)分區(qū)都有各自的主副本和從副本
- 主副本叫做leader，從副本叫做follower鬼贱，處于同步狀態(tài)的副本叫做in-sync
- 導(dǎo)致副本同步失敗的原因：
- 網(wǎng)絡(luò)擁塞導(dǎo)致復(fù)制變慢
- broker 發(fā)生奔潰導(dǎo)致復(fù)制滯后
- 持續(xù)的副本：持續(xù)請(qǐng)求得到最新消息副本被稱為同步的副本移怯。在首領(lǐng)發(fā)生失效時(shí)，只有同步副本才有可能被選為新首領(lǐng)
日志
Kafka 附帶了一個(gè)叫DumpLogSegment 的工具这难，可以用它查看片段的內(nèi)容舟误。它可以顯示每個(gè)消息的偏移量、校驗(yàn)和雁佳、魔術(shù)數(shù)字節(jié)脐帝、消息大小和壓縮算法同云。
Segment 是kafka文件存儲(chǔ)的最小單位
-
日志存放目錄 kafka_2.10-0.10.2.1/config/server.properties
kafka_log.png
進(jìn)入到日志目錄查看
kafka_tree.png
在kafka的文件存儲(chǔ)中糖权，同一個(gè)topic下有多個(gè)不同的Partition，每個(gè)partition都為一個(gè)目錄炸站，而每一個(gè)目錄又被平均分配成多個(gè)大小相等的Segment File中星澳，Segment File 包括一個(gè)日志數(shù)據(jù)文件和兩個(gè)索引文件(偏移量索引文件和消息時(shí)間戳索引文件)。
- ./kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/__consumer_offsets-0/00000000000000000000.log --print-data-log
Starting offset: 12
baseOffset: 12 lastOffset: 12 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1560145317620 size: 136 magic: 2 compresscodec: NONE crc: 692425981 isvalid: true
| offset: 12 CreateTime: 1560145317620 keysize: -1 valuesize: 66 sequence: -1 headerKeys: [] payload: {"path":"/matchingHandler.userCancel","query":{"uid":"200208305"}}
baseOffset: 13 lastOffset: 13 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 136 CreateTime: 1560237884593 size: 136 magic: 2 compresscodec: NONE crc: 2666499378 isvalid: true
| offset: 13 CreateTime: 1560237884593 keysize: -1 valuesize: 66 sequence: -1 headerKeys: [] payload: {"path":"/matchingHandler.userCancel","query":{"uid":"200208631"}}
baseOffset: 14 lastOffset: 14 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 272 CreateTime: 1560323094331 size: 136 magic: 2 compresscodec: NONE crc: 926590729 isvalid: true
| offset: 14 CreateTime: 1560323094331 keysize: -1 valuesize: 66 sequence: -1 headerKeys: [] payload: {"path":"/matchingHandler.userCancel","query":{"uid":"200208305"}}
baseOffset: 15 lastOffset: 15 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 408 CreateTime: 1560421275824 size: 136 magic: 2 compresscodec: NONE crc: 3081157511 isvalid: true
| offset: 15 CreateTime: 1560421275824 keysize: -1 valuesize: 66 sequence: -1 headerKeys: [] payload: {"path":"/matchingHandler.userCancel","query":{"uid":"200208300"}}
baseOffset: 16 lastOffset: 16 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 544 CreateTime: 1560426875868 size: 136 magic: 2 compresscodec: NONE crc: 101509911 isvalid: true
Segment文件命名規(guī)則：Partition全局的第一個(gè)Segment從0開始旱易，后續(xù)每個(gè)Segment文件名為上一個(gè)Segment文件最后一條消息的offset值禁偎。數(shù)值最大為64位long型，19位數(shù)字字符長(zhǎng)度阀坏，沒有數(shù)字用0填充如暖。
| 關(guān)鍵字 | 解釋 |
|---|---|
| offset | 消息在partition中的絕對(duì)offset。能表示這是partition的第多少條消息 |
| message | message大小 |
| CRC32 | 用crc32校驗(yàn)message |
| magic | 表示本次發(fā)布kafka服務(wù)程序協(xié)議版本號(hào) |
| attributes | 表示為獨(dú)立版本忌堂、或標(biāo)識(shí)壓縮類型盒至、或編碼類型 |
| key length | 表示key的長(zhǎng)度，當(dāng)key為-1時(shí)士修，K byte key字段不填 |
| key | 可選 |
| value bytes payload | 實(shí)際消息數(shù)據(jù) |
index 文件的存儲(chǔ)方式
- index 文件是二進(jìn)制存儲(chǔ)的枷遂，每條索引都記錄了消息的相對(duì)offset和在文件中的物理位置。這里的相對(duì)offset和log文件里的offset不同棋嘲，相對(duì)offset是每個(gè)segment都從1開始的酒唉，而絕對(duì)offset在整個(gè)partition中都是唯一的。
分段策略
| 屬性名 | 含義 | 默認(rèn)值 |
|---|---|---|
| log.roll.{hours,ms} | 日志滾動(dòng)的周期時(shí)間沸移，到達(dá)指定周期時(shí)間時(shí)痪伦，強(qiáng)制生成一個(gè)新的segment | 168(7 day) |
| log.segment.bytes | 每個(gè)segment的最大容量。到達(dá)指定容量時(shí)雹锣，將強(qiáng)制生成一個(gè)新的segment | 1G(-1 為不限制) |
| log.retention.check.interval.ms | 日志片段文件檢查的周期時(shí)間 | 60000 |
日志刷新策略
Kafka的日志實(shí)際上是開始是在緩存中的网沾，然后根據(jù)一定策略定期一批一批寫入到日志文件中去，以提高吞吐量.
| 屬性名 | 含義 | 默認(rèn)值 |
|---|---|---|
| log.flush.interval.messages | 消息達(dá)到多少條時(shí)將數(shù)據(jù)寫入到日志文件 | 10000 |
| log.flush.interval.ms | 當(dāng)達(dá)到該時(shí)間時(shí)笆制，強(qiáng)制執(zhí)行一次flush | null |
| log.flush.shceduler.interval.ms | 周期性檢查绅这，是否需要將信息flush | 暫時(shí)沒有找到 |
集群成員關(guān)系
- Kafka 使用Zookeeper來維護(hù)集群成員的信息。每個(gè)broker都有一個(gè)唯一標(biāo)識(shí)符在辆，這個(gè)標(biāo)識(shí)符可以在配置文件里指定证薇，也可以自動(dòng)生成度苔。在broker啟動(dòng)的時(shí)候，它通過創(chuàng)建臨時(shí)節(jié)點(diǎn)把自己的ID注冊(cè)到Zookeeper浑度。Kafka組件訂閱Zookeeper的 /brokers/ids 路徑(broker在Zookeeper上的注冊(cè)路徑)寇窑，當(dāng)有broker加入集群或退出集群時(shí)，這些組件就可以獲得通知箩张。在broker停機(jī)甩骏、出現(xiàn)網(wǎng)絡(luò)分區(qū)或長(zhǎng)時(shí)間垃圾回收停頓時(shí)，broker會(huì)從Zookeeper上斷開連接先慷，此時(shí)broker在啟動(dòng)時(shí)創(chuàng)建的臨時(shí)節(jié)點(diǎn)會(huì)自動(dòng)從Zookeeper上移除饮笛。監(jiān)聽broker列表的Kafka組件會(huì)被告知該broker已移除。在關(guān)閉broker時(shí)论熙，它對(duì)應(yīng)的節(jié)點(diǎn)也會(huì)消失福青，不過它的ID會(huì)繼續(xù)存在于其他數(shù)據(jù)結(jié)構(gòu)中
處理請(qǐng)求
- broker 的大部分工作是處理客戶端、分區(qū)副本和控制器發(fā)送給分區(qū)首領(lǐng)的請(qǐng)求脓诡。Kafka提供一個(gè)二進(jìn)制協(xié)議(基于TCP),指定了請(qǐng)求信息的格式以及broker如何對(duì)請(qǐng)求做出響應(yīng)——包括成功處理請(qǐng)求或在處理請(qǐng)求過程中遇到錯(cuò)誤无午。
- 客戶端發(fā)起連接并發(fā)送請(qǐng)求，broker處理請(qǐng)求并作出響應(yīng)祝谚。broker按照請(qǐng)求到達(dá)的順序來處理它們——這種順序保證讓Kafka具有了消息隊(duì)列的特性宪迟，同時(shí)保證保存的消息也是有序的。
- 標(biāo)志消息頭
| 參數(shù) | 描述 |
|---|---|
| Request type | API key |
| Request version | broker可以處理不同版本的客戶端請(qǐng)求交惯，并根據(jù)客戶端版本做出不同的響應(yīng) |
| Correlation ID | 一個(gè)具有唯一性的數(shù)字次泽，用于標(biāo)識(shí)請(qǐng)求消息，同時(shí)也會(huì)出現(xiàn)在響應(yīng)消息和錯(cuò)誤日志里(用于診斷問題) |
| Client ID | 用于標(biāo)識(shí)發(fā)送請(qǐng)求的客戶端 |
- broker會(huì)在它所監(jiān)聽的每一個(gè)端口上運(yùn)行一個(gè)Acceptor線程商玫，這個(gè)線程會(huì)創(chuàng)建一個(gè)連接箕憾，并把它交給Processor線程去處理。Processor線程("網(wǎng)絡(luò)線程")的數(shù)量是可以配置的 拳昌。網(wǎng)絡(luò)線程負(fù)責(zé)從客戶端獲取請(qǐng)求信息袭异，把它們放進(jìn)請(qǐng)求隊(duì)列，然后從響應(yīng)隊(duì)列獲取響應(yīng)信息炬藤，把它們發(fā)送給客戶端御铃。
- 客戶端如何知道往哪里發(fā)送請(qǐng)求
- 客戶端使用了另一種請(qǐng)求類型，也就是元數(shù)據(jù)請(qǐng)求沈矿。這種請(qǐng)求包含了客戶端感興趣的主題列表上真。服務(wù)器端的響應(yīng)消息里指明了這些主題所包含的分區(qū)、每個(gè)分區(qū)都有哪些副本羹膳，以及哪個(gè)副本是首領(lǐng)睡互。元數(shù)據(jù)請(qǐng)求可以發(fā)送給任意一個(gè)broker，因?yàn)樗衎roker都緩存了這些信息。
控制器
- 控制器其實(shí)就是一個(gè)broker就珠。集群里第一個(gè)啟動(dòng)的broker通過在Zookeeper里創(chuàng)建一個(gè)臨時(shí)節(jié)點(diǎn) /controller 讓自己成為控制器寇壳。其他broker在啟動(dòng)時(shí)也會(huì)嘗試創(chuàng)建這個(gè)節(jié)點(diǎn)，不過它們會(huì)收到一個(gè)"節(jié)點(diǎn)已存在"的異常妻怎，然后"意識(shí)"到控制器節(jié)點(diǎn)已存在壳炎，也就是說集群里已經(jīng)有一個(gè)控制器了。其他broker在控制器節(jié)點(diǎn)上創(chuàng)建Zookeeper watch對(duì)象逼侦，這樣它們就可以收到這個(gè)節(jié)點(diǎn)的變更通知匿辩。這種方式可以確保集群里一次只有一個(gè)控制器存在。
- 如果控制器被關(guān)閉或者Zookeeper斷開連接榛丢，Zookeeper上的臨時(shí)節(jié)點(diǎn)就會(huì)消失铲球。集群里的其他broker通過watch對(duì)象得到控制器節(jié)點(diǎn)消失的通知，它們會(huì)嘗試讓自己成為新的控制器涕滋。第一個(gè)在Zookeeper里成功成功創(chuàng)建控制器節(jié)點(diǎn)的broker就會(huì)成為新的控制器睬辐，其他節(jié)點(diǎn)會(huì)收到"節(jié)點(diǎn)已存在"的異常挠阁，然后在新的控制器節(jié)點(diǎn)上再次創(chuàng)建watch對(duì)象宾肺。每個(gè)新選出的控制器通過Zookeeper的條件遞增操作獲得一個(gè)全新的、數(shù)值更大的controller epoch侵俗。其他broker在知道當(dāng)前controller epoch后锨用，如果收到有控制器發(fā)出的包含舊epoch的消息，就會(huì)忽略它們隘谣。
- 當(dāng)控制器發(fā)現(xiàn)一個(gè)broker已經(jīng)離開集群(通過觀察相關(guān)的Zookeeper路徑)增拥，它就知道，那些失去首領(lǐng)的分區(qū)需要一個(gè)新首領(lǐng)(這些分區(qū)的首領(lǐng)剛好在這個(gè)broker上)寻歧≌普ぃ控制器遍歷這些分區(qū)，并確定誰(shuí)應(yīng)該成為新首領(lǐng)(簡(jiǎn)單來說就是分區(qū)副本列表里的下一個(gè)副本)码泛，然后向所有包含新首領(lǐng)或現(xiàn)有跟隨者的broker發(fā)送請(qǐng)求猾封。該請(qǐng)求消息包含了誰(shuí)是新首領(lǐng)已經(jīng)誰(shuí)是分區(qū)跟隨者的信息。隨后新首領(lǐng)開始處理來著生產(chǎn)者和消費(fèi)者的請(qǐng)求噪珊，而跟隨者開始從新首領(lǐng)那里復(fù)制消息晌缘。
- 當(dāng)控制器發(fā)現(xiàn)一個(gè)broker加入集群時(shí)，它會(huì)使用broker ID來檢查新加入的broker是否包含現(xiàn)有的分區(qū)副本痢站。如果有磷箕，控制器就把變更通知發(fā)送給新加入的broker和其他broker，新broker上的副本開始從首領(lǐng)那里復(fù)制消息阵难。
Kafka 消費(fèi)過程分析
- Kafka提供了兩套consumer API：高級(jí)Consumer API 和 低級(jí) Consumer API
- 高級(jí)API 優(yōu)點(diǎn)
- 高級(jí)API寫起來簡(jiǎn)單
- 不需要自行去管理offset岳枷，系統(tǒng)通過zookeeper自行管理
- 不需要管理分區(qū)，副本等情況，系統(tǒng)自動(dòng)管理
- 消費(fèi)者斷線會(huì)自動(dòng)根據(jù)上一次記錄在zookeeper中的offset去接著獲取數(shù)據(jù)(默認(rèn)設(shè)置1分鐘更新一下zookeeper中存的offset)
- 可以使用group來區(qū)分對(duì)同一個(gè)topic的不同程序訪問分離開來(不同的group記錄不同的offset空繁，這樣不同程序讀取同一個(gè)topic才不會(huì)因?yàn)閛ffset互相影響)
- 高級(jí)API 缺點(diǎn)
- 不能自行控制offset(對(duì)于某些特殊需求)
- 不能細(xì)化控制氢烘，如分區(qū)、副本家厌、zk等
- 低級(jí)API
- 低級(jí)API優(yōu)點(diǎn)
- 能夠讓開發(fā)者自己控制offset播玖，想從哪里讀取就從哪里讀取
- 自行控制連接分區(qū)，對(duì)分區(qū)自定義進(jìn)行負(fù)載均衡
- 對(duì)zookeeper的依賴性降低(如：offset不一定非要靠zk存儲(chǔ)饭于，自行存儲(chǔ)offset即可蜀踏，比如存儲(chǔ)在文件或則內(nèi)存中)
- 低級(jí)API缺點(diǎn)
- 太過復(fù)雜，需要自行控制offset掰吕，連接哪個(gè)分區(qū)果覆，找到分區(qū)leader等
kafka復(fù)制原理
Kafka的復(fù)制機(jī)制和分區(qū)的多副本架構(gòu)是Kafka可靠性保證的核心。把消息寫入多個(gè)副本可以是Kafka在發(fā)送奔潰時(shí)仍能保證消息的持久性殖熟。
消費(fèi)的發(fā)送方式：主題\value局待、主題\key\value、主題\分區(qū)\key\value菱属、主題\分區(qū)\時(shí)間戳
\key\valueKafka 中topic的每個(gè)partition有一個(gè)預(yù)寫式的日志文件钳榨，雖然partition可以繼續(xù)細(xì)分為若干個(gè)segment文件，但是對(duì)于上層應(yīng)用來說可以將partition看成最小的存儲(chǔ)單元纽门，每個(gè)partition都由一些列有序的薛耻、不可變的消息組成，這些消息被連續(xù)的追加到partition中赏陵。
- LEO：LogEndOffset的縮寫饼齿，表示每個(gè)partition的log最后一條Message的位置
- HW： 是HighWatermark的縮寫，是指consumer能夠看到的此partition的位置
- 具體描述：Kafka每個(gè)topic的partition有N個(gè)副本(replicas).
- kafka 通過多副本機(jī)制實(shí)現(xiàn)故障自動(dòng)轉(zhuǎn)移蝙搔，當(dāng)kafka集群中一個(gè)broker失效情況下仍然保證服務(wù)可用缕溉。kafka中發(fā)生復(fù)制時(shí)確保partition的日志能有序地寫到其他節(jié)點(diǎn)上，N個(gè)replicas中吃型，其中一個(gè)replicas為leader证鸥，其他都為follower，leader處理partition的所有讀寫請(qǐng)求败玉，于此同時(shí)敌土，follower會(huì)被動(dòng)定期地去復(fù)制leader的數(shù)據(jù)。kafka提供了數(shù)據(jù)復(fù)制算法保證运翼，如果leader發(fā)生故障或掛掉返干，一個(gè)新leader被選舉并接受客戶端的消息成功寫入。
- leader負(fù)責(zé)維護(hù)和跟蹤ISR中所有follower滯后的狀態(tài).
- 當(dāng)producer發(fā)送一條消息到broker后血淌，leader寫入消息并復(fù)制到所有follower矩欠。消息提交之后才被成功復(fù)制到所有的同步副本财剖。消息復(fù)制延遲受最慢的follower限制，重要的是快速檢測(cè)慢副本癌淮，如果follower"落后"太多或者失效躺坟，leader將會(huì)把它從ISR中刪除.
leader 將某個(gè)follower提出ISR列表的情況：
- 按數(shù)量——如果leader當(dāng)前的offset已經(jīng)到10，但是某個(gè)follower同步的數(shù)據(jù)還是2乳蓄，但是kafka對(duì)于數(shù)量的偏差設(shè)置為6咪橙。如果當(dāng)前偏差小于等于設(shè)置的偏差，那么會(huì)將該follower提出ISR列表虚倒，進(jìn)入到OSR列表[所有的副本數(shù)據(jù) = ISR + OSR]
- 按時(shí)間——有新數(shù)據(jù)美侦，多久沒有發(fā)送確認(rèn)信息
- hwpng.png
ISR(副本同步隊(duì)列)
- ISR 是所有副本的一個(gè)子集，由leader維護(hù)ISR列表魂奥，follower從leader同步數(shù)據(jù)有一些延遲菠剩，包括延遲時(shí)間 replica.lag.time.max.ms和延遲條數(shù)replica.lag.max.messages兩個(gè)維度，當(dāng)前最新的版本0.10.x中只支持replica.lag.time.max.ms這個(gè)維度）耻煤。任意一個(gè)超過閾值都會(huì)把follower剔除出ISR具壮，存入OSR(Outof-Sync Replicas)列表，新加入的follower也會(huì)先存放在OSR中哈蝇。
- leader 新寫入的信息棺妓，consumer不能立刻消費(fèi)，leader會(huì)等待該消息被所有ISR中的replicas同步后更新HW买鸽，此時(shí)消息才能被consumer消費(fèi)涧郊。這樣就保證了如果leader所在的broker失效，該消息仍然可以從新選舉的leader中獲取眼五。對(duì)于來自內(nèi)部的broker的讀取請(qǐng)求，沒有HW的限制彤灶。
- 同步復(fù)制要求所有的能工作的follower都復(fù)制完看幼，這條消息才會(huì)被commit，這種復(fù)制方式是否極大的影響了吞吐率幌陕？
- 異步復(fù)制方式
- follower異步的從leader復(fù)制數(shù)據(jù)诵姜，數(shù)據(jù)只要被leader寫入log就被認(rèn)為已經(jīng)commit，這種情況下如果follower都還沒有復(fù)制完搏熄，落后于leader時(shí)棚唆，突然leader宕機(jī)，則會(huì)丟失數(shù)據(jù)心例。而kafka的這種使用ISR的方式則很好的均衡了確保數(shù)據(jù)不丟失已經(jīng)吞吐率宵凌。
- Kafka的管理最終都會(huì)反饋到Zookeeper節(jié)點(diǎn)上。
- 具體位置：/brokers/topics/[topic]/partitions/[partition]/state.
- 目前有兩個(gè)地方會(huì)對(duì)這個(gè)Zookeeper的節(jié)點(diǎn)進(jìn)行維護(hù)：
- Controller維護(hù)：Controller 下的LeaderSelector會(huì)選舉新的leader止后，ISR和新的leader_epoch及controller_epoch寫入Zookeeper的相關(guān)節(jié)點(diǎn)中瞎惫。同時(shí)發(fā)起LeaderAndIsrRequest通知所有的replicas溜腐。
- leader維護(hù)：leader有單獨(dú)的線程定期檢測(cè)ISR中follower是否脫離ISR，如果發(fā)現(xiàn)ISR變化瓜喇，則會(huì)將新的ISR的信息返回到Zookeeper的相關(guān)節(jié)點(diǎn)中挺益。
- Kafka集群中的其中一個(gè)Broker會(huì)被選舉為Controller，主要負(fù)責(zé)Partition管理和副本狀態(tài)管理乘寒，也會(huì)執(zhí)行類似于重分配partition之類的管理任務(wù)望众。
kafka 數(shù)據(jù)可靠性
- 數(shù)據(jù)丟失的可能:可以采用callback的方式進(jìn)行處理，判斷異常信息是否為空伞辛，如果為空表示正常發(fā)送了黍檩，否則就有異常，可進(jìn)行特殊處理
- 當(dāng)producer向leader發(fā)送數(shù)據(jù)時(shí)始锚，可以通過acks參數(shù)來設(shè)置數(shù)據(jù)可靠性的級(jí)別：
- 1(默認(rèn))：這意味著producer在ISR中的leader已成功收到的數(shù)據(jù)并得到確認(rèn)后發(fā)送下一條message刽酱。如果leader宕機(jī)了，則會(huì)丟失數(shù)據(jù)瞧捌。
- 0：這意味著producer無需等待來自broker的確認(rèn)而繼續(xù)發(fā)送下一批消息棵里。這種情況下數(shù)據(jù)傳輸效率最高，但是數(shù)據(jù)可靠性是最低的
- all：leader需要等待所有備份都寫入日志姐呐，這種策略會(huì)保證只要有一個(gè)備份存活就不會(huì)丟失數(shù)據(jù)殿怜，這是最強(qiáng)的保證。
kafka 消息傳輸保障
- Kafka確保消息在producer和consumer之間傳輸曙砂。有以下三種可能的傳輸保障
- At most once : 消息可能丟失头谜，但絕不會(huì)重復(fù)傳輸
- At least once : 消息絕不會(huì)丟，但可能重復(fù)傳輸
- Exactly once: 每條消息肯定會(huì)被傳輸一次且僅傳輸一次
kafka leader 和 follower 如何通信
疑問
- 一個(gè)broker服務(wù)下鸠澈，是否可以創(chuàng)建多個(gè)分區(qū)柱告？
- 可以，broker數(shù)與分區(qū)數(shù)沒有關(guān)系
- 一個(gè)broker服務(wù)下笑陈，是否可以創(chuàng)建多個(gè)副本因子?
- 不可以际度，會(huì)報(bào)錯(cuò);
創(chuàng)建主題時(shí)，副本因子應(yīng)該小于等于可用的broker數(shù)Error while executing topic command : replication factor: 3 larger than available brokers: 1 [2019-07-23 17:34:45,963] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: replication factor: 3 larger than available brokers: 1 (kafka.admin.TopicCommand$)
- 不可以际度，會(huì)報(bào)錯(cuò);
- 在kafka中涵妥，每一個(gè)分區(qū)會(huì)有一個(gè)編號(hào)乖菱，從0開始
- 當(dāng)執(zhí)行刪除命令之后，topic不是物理刪除蓬网，而是一個(gè)標(biāo)記刪除的操作.
- 標(biāo)記刪除之后的主題是否還可以繼續(xù)生產(chǎn)數(shù)據(jù)窒所？
- 不會(huì)有影響
- 如何保證一個(gè)主題下的數(shù)據(jù)，一定是有序的(生產(chǎn)與消費(fèi)的順序一致)
- 讓主題下只有一個(gè)分區(qū)
- 某一個(gè)主題下的分區(qū)數(shù)帆锋，對(duì)于消費(fèi)組來說吵取，應(yīng)該小于等于該主題下的分區(qū)數(shù)。
- 在使用kafka的過程中窟坐，如何保證數(shù)據(jù)的不丟失海渊，不重復(fù)的問題绵疲？
- 如何確保Producer不丟失數(shù)據(jù)?
- ACK (應(yīng)答機(jī)制設(shè)置為2)
- Kafka 的用途？使用場(chǎng)景臣疑？
- 消息系統(tǒng)盔憨；實(shí)時(shí)監(jiān)控或者離線處理；日志收集
- 異常處理讯沈、日常系統(tǒng)削峰郁岩、解耦、提速缺狠、廣播
- Kafka中的ISR问慎、AR代表什么？ISR的伸縮?
- ISR: In-Sync Replicas 副本同步隊(duì)列
- AR: Assigned Replicas 所有副本
- ISR是由leader維護(hù)挤茄，follower從leader同步數(shù)據(jù)有一些延遲(包括延遲時(shí)間replica.lag.time.max.ms 和 延遲條數(shù) replica.lag.max.message兩個(gè)維度如叼，當(dāng)前最新的版本0.10.x 中只支持replica.lag.time.max.ms這個(gè)維度),任意一個(gè)超過閾值都會(huì)把follower剔除出ISR,存入OSR(Outof-Sync Replicas)列表，新加入的follower也會(huì)存放在OSR中穷劈。AR=ISR+OSR
- Kafka中的HW笼恰、LEO、LSO歇终、LW等分別代表什么社证？
- HW: High Watermark 高水位，取一個(gè)partition對(duì)應(yīng)的ISR中最小的LEO作為HW评凝，consumer最多只能消費(fèi)到HW所在的位置上一條信息
- LEO: LogEndOffset 當(dāng)然日志文件中下一條代寫信息的offset
- HW/LEO 這兩個(gè)都是指最后一條的下一條的位置而不是最后一條的位置
- LSO: Last Stable Offset 對(duì)未完成的事務(wù)而言追葡，LSO的值等于事務(wù)中第一條消息的位置(firstUnstableOffset),對(duì)已完成的事務(wù)而言，它的值同HW相同
- LW: Low Watermark 低水位奕短，代表AR集合中最小的logStartOffset值
- Kafka 中是怎么體現(xiàn)消息順序性的?
- Kafka每個(gè)partition中的消息在寫入是都是有序的宜肉，消費(fèi)時(shí)，每個(gè)partition只能被每一個(gè)group中的消費(fèi)者消費(fèi)篡诽，保證了消費(fèi)時(shí)也是有序的
- 整個(gè)topic不保證有序崖飘。如果為了保證topic整個(gè)有序，那么將partition調(diào)整為1
- Kafka中的分區(qū)器杈女、序列化器、攔截器之間的處理順序是什么?
- 攔截器 -> 序列化器 -> 分區(qū)器
- Kafka 生產(chǎn)者客戶端中使用了幾個(gè)線程來處理吊圾？
- 2個(gè)达椰，主線程和Sender線程。主線程負(fù)責(zé)創(chuàng)建消息项乒，然后通過分區(qū)器啰劲、序列化器、攔截器作用之后緩存到累加器RecordAccumulator中檀何。Sender線程負(fù)責(zé)將RecordAccumulator中消息發(fā)送到Kafka中
- 消費(fèi)者提交消費(fèi)位移時(shí)提交的是當(dāng)前消費(fèi)到的最新消息的offset還是offset+1蝇裤？
- offset + 1
- 造成重復(fù)消費(fèi)的原因：
- 消費(fèi)者消費(fèi)后沒有commit offset(程序奔潰/強(qiáng)行kill/消費(fèi)耗時(shí)/自動(dòng)提交偏移情況下unscrible)
- 造成消息漏消費(fèi)的原因:
- 消費(fèi)者沒有處理完消息廷支，提交offset(自動(dòng)提交偏移，未處理情況下程序異常結(jié)束)
- KafkaConsumer 是非線程安全的栓辜，如何實(shí)現(xiàn)多線程消費(fèi)
- 在每個(gè)線程中創(chuàng)建一個(gè)KafkaConsumer
- 單線程創(chuàng)建KafkaConsumer恋拍，多個(gè)處理線程處理消息
- 消費(fèi)者與消費(fèi)組之間的關(guān)系
- 消費(fèi)者從屬于消費(fèi)組，消費(fèi)偏移以消費(fèi)組為單位藕甩。每個(gè)消費(fèi)組可以獨(dú)立消費(fèi)主題的所有數(shù)據(jù)施敢，同一消費(fèi)組內(nèi)消費(fèi)者共同消費(fèi)主題數(shù)據(jù)，每個(gè)分區(qū)只能被同一消費(fèi)組內(nèi)一個(gè)消費(fèi)者消費(fèi)
- 使用kafka-topics.sh 創(chuàng)建(刪除)了一個(gè)topic之后狭莱，kafka背后執(zhí)行了什么邏輯
- 創(chuàng)建：在zk上 /brokers/topics/下節(jié)點(diǎn) Kafka broker 會(huì)監(jiān)聽節(jié)點(diǎn)變化創(chuàng)建主題
- 刪除: 調(diào)用腳本刪除topic會(huì)在zk上將topic設(shè)置待刪除標(biāo)志僵娃，kafka后臺(tái)有定時(shí)線程會(huì)掃描所有需要?jiǎng)h除的topic進(jìn)行刪除
- 創(chuàng)建topic時(shí)如何選擇合適的分區(qū)數(shù)
- 根據(jù)集群的機(jī)器數(shù)量和需要的吞吐量來決定適合的分區(qū)數(shù)
- Kafka 目前有哪些內(nèi)部topic，特征腋妙，作用
- __consumer_offsets 保證消費(fèi)組的偏移
- 優(yōu)先副本是什么默怨？有什么特殊作用
- 默認(rèn)的leader副本
- 發(fā)送leader變化時(shí)重選舉會(huì)優(yōu)先選擇優(yōu)先副本作為leader
- Kafka的Log Retention的理解
- kafka 留存策略包括刪除和壓縮兩種
- 刪除：根據(jù)時(shí)間和大小兩種方式進(jìn)行刪除，大小是整個(gè)partition日志文件的大小骤素，超過的會(huì)從老到新依次刪除匙睹；時(shí)間指定日志文件中最大時(shí)間戳而非文件的最后修改時(shí)間
- 壓縮：相同key的value只保存一個(gè) 壓縮過的是clean 未壓縮的dirty 壓縮之后的偏移量不連續(xù) 未壓縮時(shí)連續(xù)
