[[277321]]

首先從kafka如何創(chuàng)建一個(gè)topic來(lái)開(kāi)始：

kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test 

其中有這么幾個(gè)參數(shù)：

• --zookeeper：zookeeper的地址
• --replication-factor：副本因子
• --partitions：分區(qū)個(gè)數(shù)(默認(rèn)是1)
• --topic：topic名稱

二.什么是分區(qū)

一個(gè)topic可以有多個(gè)分區(qū)，每個(gè)分區(qū)的消息都是不同的。 雖然分區(qū)可以提供更高的吞吐量，但是分區(qū)不是越多越好。一般分區(qū)數(shù)不要超過(guò)kafka集群的機(jī)器數(shù)量。分區(qū)越多占用的內(nèi)存和文件句柄。 一般分區(qū)設(shè)置為3-10個(gè)。比如現(xiàn)在集群有3個(gè)機(jī)器，要?jiǎng)?chuàng)建一個(gè)名為test的topic，分區(qū)數(shù)為2，那么如圖：

partiton都是有序切順序不可變的記錄集，并且不斷追加到log文件，partition中的每一個(gè)消息都回分配一個(gè)id，也就是offset(偏移量)，offset用來(lái)標(biāo)記分區(qū)的一條記錄 ，這里就用官網(wǎng)的圖了，我畫(huà)的不好：

2.1 producer端和分區(qū)關(guān)系

就圖上的情況，producer端會(huì)把mq給哪個(gè)分區(qū)呢?這也是上一節(jié)我們提到的一個(gè)參數(shù)partitioner.class。 默認(rèn)分區(qū)器的處理是：有key則用murmur2算法計(jì)算key的哈希值，對(duì)總分區(qū)取模算出分區(qū)號(hào)，無(wú)key則輪詢。(org.apache.kafka.clients.producer.internals.DefaultPartitioner#partition)。當(dāng)然了我們也可以自定義分區(qū)策略，只要實(shí)現(xiàn)org.apache.kafka.clients.producer.Partitioner接口即可：

/** 
 * Compute the partition for the given record. 
 * 
 * @param topic The topic name 
 * @param key The key to partition on (or null if no key) 
 * @param keyBytes serialized key to partition on (or null if no key) 
 * @param value The value to partition on or null 
 * @param valueBytes serialized value to partition on or null 
 * @param cluster The current cluster metadata 
 */ 
 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
 List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
 int numPartitions = partitions.size(); 
 if (keyBytes == null) { 
 int nextValue = nextValue(topic); 
 List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); 
 if (availablePartitions.size() > 0) { 
 int part = Utils.toPositive(nextValue) % availablePartitions.size(); 
 return availablePartitions.get(part).partition(); 
 } else { 
 // no partitions are available, give a non-available partition 
 return Utils.toPositive(nextValue) % numPartitions; 
 } 
 } else { 
 // hash the keyBytes to choose a partition 
 return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; 
 } 
 } 

2.2 consumer端和分區(qū)關(guān)系

先來(lái)看下官網(wǎng)對(duì)于消費(fèi)組的定義：Consumers label themselves with a consumer group name, and each record published to a topic is delivered to one consumer instance within each subscribing consumer group.

翻譯：消費(fèi)者使用一個(gè)消費(fèi)者組名來(lái)標(biāo)記自己，一個(gè)topic的消息會(huì)被發(fā)送到訂閱它的消費(fèi)者組的 一個(gè) 消費(fèi)者實(shí)例上。

consumer group是用于實(shí)現(xiàn)高伸縮性，高容錯(cuò)性的consumer機(jī)制。如果有consumer掛了或者新增一個(gè)consumer，consumer group會(huì)進(jìn)行重平衡(rebalance)，重平衡機(jī)制會(huì)在consumer篇具體講解，本節(jié)不講。那么按照上面的圖繼續(xù)畫(huà)消費(fèi)者端：

這里是最好的情況，2個(gè)partition對(duì)應(yīng)1個(gè)group中的2個(gè)consumer。那么思考，如果一個(gè)消費(fèi)組的消費(fèi)者大于分區(qū)數(shù)呢?或者小于分區(qū)數(shù)呢?

如果一個(gè)消費(fèi)組的消費(fèi)者大于分區(qū)數(shù)，那么相當(dāng)于多余的消費(fèi)者是一種浪費(fèi)，多余的消費(fèi)者將無(wú)法消費(fèi)消息。

如果一個(gè)消費(fèi)組的消費(fèi)者小于分區(qū)數(shù)，會(huì)有對(duì)應(yīng)的消費(fèi)者分區(qū)分配策略。一種是Range(默認(rèn))，一種是RoundRobin(輪詢)，當(dāng)然也可以自定義策略。 其實(shí)思想換湯不換藥的啊，每個(gè)消費(fèi)者能負(fù)載均衡的工作。 具體會(huì)在消費(fèi)者篇講解，這里不講。

建議：配置分區(qū)數(shù)是消費(fèi)者數(shù)的整數(shù)倍

三.副本與ISR設(shè)計(jì)

3.1 什么是副本

在創(chuàng)建topic的時(shí)候有個(gè)參數(shù)是--replication-factor來(lái)設(shè)定副本數(shù)。Kafka利用多份相同的備份保持系統(tǒng)的高可用性，這些備份在Kafka中被稱為副本(replica)。副本分為3類：

• leader副本：響應(yīng)producer端的讀寫(xiě)請(qǐng)求
• follower副本：備份leader副本的數(shù)據(jù)， 不響應(yīng)producer端的讀寫(xiě)請(qǐng)求!
• ISR副本集合：包含1個(gè)leader副本和所有follower副本(也可能沒(méi)有follower副本)

Kafka會(huì)把所有的副本均勻分配到kafka-cluster中的所有broker上，并從這些副本中挑選一個(gè)作為leader副本，其他成為follow副本。如果leader副本所在的broker宕機(jī)了，那么其中的一個(gè)follow副本就會(huì)成為leader副本。leader副本接收producer端的讀寫(xiě)請(qǐng)求，而follow副本只是向leader副本請(qǐng)求數(shù)據(jù)不會(huì)接收讀寫(xiě)請(qǐng)求!

3.2 副本同步機(jī)制

上面說(shuō)了ISR就是動(dòng)態(tài)維護(hù)一組同步副本集合，leader副本總是包含在ISR集合中。只有ISR中的副本才有資格被選舉為leader副本。當(dāng)producer端的ack參數(shù)配置為all(-1)時(shí)，producer寫(xiě)入的mq需要ISR所有副本都接收到，才被視為已提交。當(dāng)然了，上一節(jié)就提到了，使用ack參數(shù)必須配合broker端的min.insync.replicas(默認(rèn)是1)參數(shù)一起用才能達(dá)到效果，該參數(shù)控制寫(xiě)入isr中的多少副本才算成功。 如果ISR中的副本數(shù)少于min.insync.replicas時(shí)，客戶端會(huì)返回異常org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejected since there are fewer in-sync replicas than required。

要了解副本同步機(jī)制需要先學(xué)習(xí)幾個(gè)術(shù)語(yǔ)：

• High Watermark：副本高水位值，簡(jiǎn)稱HW， 小于HW或者說(shuō)在HW以下的消息都被認(rèn)為是“已備份的”，HW指向的也是下一條消息! leader副本的HW值決定consumer能poll的消息數(shù)量!consumer只能消費(fèi)小于HW值的消息!
• LEO：log end offset，下一條消息的位移。 也就是說(shuō)LEO指向的位置是沒(méi)有消息的!
• remote LEO：嚴(yán)格來(lái)說(shuō)這是一個(gè)集合。leader副本所在broker的內(nèi)存中維護(hù)了一個(gè)Partition對(duì)象來(lái)保存對(duì)應(yīng)的分區(qū)信息，這個(gè)Partition中維護(hù)了一個(gè)Replica列表，保存了該分區(qū)所有的副本對(duì)象。除了leader Replica副本之外，該列表中其他Replica對(duì)象的LEO就被稱為remote LEO。

下面舉個(gè)一個(gè)實(shí)際的例子(本例子參考胡夕博客)，該例子中的topic是單分區(qū)，副本因子是2。也就是說(shuō)一個(gè)leader副本，一個(gè)follower副本，ISR中包含這2個(gè)副本集合。我們首先看下當(dāng)producer發(fā)送一條消息時(shí)，leader/follower端broker的副本對(duì)象到底會(huì)發(fā)生什么事情以及分區(qū)HW是如何被更新的。首先是初始狀態(tài)：

此時(shí)producer給該topic分區(qū)發(fā)送了一條消息。此時(shí)的狀態(tài)如下圖所示：

如上圖所見(jiàn)，producer發(fā)送消息成功后(假設(shè)acks=1, leader成功寫(xiě)入即返回)，follower發(fā)來(lái)了新的FECTH請(qǐng)求，依然請(qǐng)求fetchOffset = 0的數(shù)據(jù)。和上次不同的是，這次是有數(shù)據(jù)可以讀取的，因此整個(gè)處理流程如下圖：

顯然，現(xiàn)在leader和follower都保存了位移是0的這條消息，但兩邊的HW值都沒(méi)有被更新，它們需要在下一輪FETCH請(qǐng)求處理中被更新，如下圖所示：

簡(jiǎn)單解釋一下， 第二輪FETCH請(qǐng)求中，follower發(fā)送fetchOffset = 1的FETCH請(qǐng)求——因?yàn)閒etchOffset = 0的消息已經(jīng)成功寫(xiě)入follower本地日志了，所以這次請(qǐng)求fetchOffset = 1的數(shù)據(jù)了。Leader端broker接收到FETCH請(qǐng)求后首先會(huì)更新other replicas中的LEO值，即將remote LEO更新成1，然后更新分區(qū)HW值為1——具體的更新規(guī)則參見(jiàn)上面的解釋。做完這些之后將當(dāng)前分區(qū)HW值(1)封裝進(jìn)FETCH response發(fā)送給follower。Follower端broker接收到FETCH response之后從中提取出當(dāng)前分區(qū)HW值1，然后與自己的LEO值比較，從而將自己的HW值更新成1，至此完整的HW、LEO更新周期結(jié)束。

3.3 ISR維護(hù)

在0.9.0.0版本之后，只有一個(gè)參數(shù)：replica.lag.time.max.ms來(lái)判定該副本是否應(yīng)該在ISR集合中，這個(gè)參數(shù)默認(rèn)值為10s。意思是如果一個(gè)follower副本響應(yīng)leader副本的時(shí)間超過(guò)10s，kafka會(huì)認(rèn)為這個(gè)副本走遠(yuǎn)了從同步副本列表移除。

四.日志設(shè)計(jì)

Kafka的每個(gè)主題相互隔離，每個(gè)主題可以有一個(gè)或者多個(gè)分區(qū)，每個(gè)分區(qū)都有記錄消息數(shù)據(jù)的日志文件：

圖中有個(gè)demo-topic的主題，這個(gè)topic有8個(gè)分區(qū)，每一個(gè)分區(qū)都存在[topic-partition]命名的消息日志文件 。在分區(qū)日志文件中，可以看到前綴一樣，但是文件類型不一樣的幾個(gè)文件。比如圖中的3個(gè)文件，(00000000000000000000.index、00000000000000000000.timestamp、00000000000000000000.log)。這稱之為一個(gè)LogSegment(日志分段)。

4.1 LogSegment

以一個(gè)測(cè)試環(huán)境的具體例子來(lái)講，一個(gè)名為ALC.ASSET.EQUITY.SUBJECT.CHANGE的topic，我們看partition0的日志文件：

每一個(gè)LogSegment都包含一些文件名一致的文件集合。文件名的固定是20位數(shù)字，如果文件名是00000000000000000000代表當(dāng)前LogSegment的第一條消息的offset(偏移量)為0，如果文件名是00000000000000000097代表當(dāng)前LogSegment的第一條消息的offset(偏移量)為97。日志文件有多種后綴的文件，重點(diǎn)關(guān)注.index、.timestamp、.log三種類型文件即可。

• .index：偏移量索引文件
• .timeindex：時(shí)間索引文件
• .log：日志文件
• .snapshot：快照文件
• .swap：Log Compaction之后的臨時(shí)文件

4.2 索引與日志文件

kafka有2種索引文件，第一種是offset(偏移量)索引文件，也就是.index結(jié)尾的文件。第二種是時(shí)間戳索引文件，也就是.timeindex結(jié)尾的文件。

我們可以用kafka-run-class.sh來(lái)查看offset(偏移量)索引文件的內(nèi)容：

可以看到每一行都是offset：xxx position：xxxx。 這兩者沒(méi)有直接關(guān)系。

• offset：相對(duì)偏移量
• position：物理地址

那么第一行的offset：12 position：4423是什么意思呢?它代表偏移量從0-12的消息的物理地址在0-4423。

同理第二行的offset：24 position：8773的意思也能猜得出來(lái)：它代表偏移量從13-24的消息的物理地址在4424-8773。

我們可以再用kafka-run-class.sh來(lái)看下.log文件的文件內(nèi)容，關(guān)注里面的baseOffset和postion的值。你看看和上面說(shuō)的對(duì)應(yīng)的上嗎。

4.3 如何用offset查找

按上面的例子，如何查詢偏移量為60的消息

根據(jù)offset首先找到對(duì)應(yīng)的LogSegment，這里找到00000000000000000000.index

通過(guò)二分法找到不大于offset的最大索引項(xiàng)，這里找到offset：24 position：8773

打開(kāi)00000000000000000000.log文件，從position為8773的那個(gè)地方開(kāi)始順序掃描直到找到offset=60的消息