Kafka 原理以及分區分配策略剖析
一、簡介
流處理平臺有以下3個特性：
- 可以讓你發布和訂閱流式的記錄。這一方面與消息隊列或者企業消息系統類似。
- 可以儲存流式的記錄，并且有較好的容錯性。
- 可以在流式記錄產生時就進行處理。
1.1 消息隊列的兩種模式
1.1.1 點對點模式
生產者將消息發送到queue中，然后消費者從queue中取出并且消費消息。消息被消費以后，queue中不再存儲，所以消費者不可能消費到已經被消費的消息。Queue支持存在多個消費者，但是對一個消息而言，只能被一個消費者消費。
1.1.2 發布/訂閱模式
生產者將消息發布到topic中，同時可以有多個消費者訂閱該消息。和點對點方式不同，發布到topic的消息會被所有訂閱者消費。
1.2 Kafka 適合什么樣的場景
它可以用于兩大類別的應用：
- 構造實時流數據管道，它可以在系統或應用之間可靠地獲取數據。(相當于message queue)。
- 構建實時流式應用程序，對這些流數據進行轉換或者影響。(就是流處理，通過kafka stream topic和topic之間內部進行變化)。
為了理解Kafka是如何做到以上所說的功能，從下面開始，我們將深入探索Kafka的特性。
首先是一些概念：
- Kafka作為一個集群，運行在一臺或者多臺服務器上。
- Kafka 通過 topic 對存儲的流數據進行分類。
- 每條記錄中包含一個key，一個value和一個timestamp（時間戳）。
1.3 主題和分區
Kafka的消息通過主題（Topic）進行分類，就好比是數據庫的表，或者是文件系統里的文件夾。主題可以被分為若干個分區（Partition），一個分區就是一個提交日志。消息以追加的方式寫入分區，然后以先進先出的順序讀取。注意，由于一個主題一般包含幾個分區，因此無法在整個主題范圍內保證消息的順序，但可以保證消息在單個分區內的順序。主題是邏輯上的概念，在物理上，一個主題是橫跨多個服務器的。
Kafka 集群保留所有發布的記錄（無論他們是否已被消費），并通過一個可配置的參數——保留期限來控制（可以同時配置時間和消息大小，以較小的那個為準）。舉個例子， 如果保留策略設置為2天，一條記錄發布后兩天內，可以隨時被消費，兩天過后這條記錄會被拋棄并釋放磁盤空間。
有時候我們需要增加分區的數量，比如為了擴展主題的容量、降低單個分區的吞吐量或者要在單個消費者組內運行更多的消費者（因為一個分區只能由消費者組里的一個消費者讀取）。從消費者的角度來看，基于鍵的主題添加分區是很困難的，因為分區數量改變，鍵到分區的映射也會變化，所以對于基于鍵的主題來說，建議在一開始就設置好分區，避免以后對其進行調整。
（注意：不能減少分區的數量，因為如果刪除了分區，分區里面的數據也一并刪除了，導致數據不一致。如果一定要減少分區的數量，只能刪除topic重建）
1.4 生產者和消費者
生產者（發布者）創建消息，一般情況下，一個消息會被發布到一個特定的主題上。生產者在默認情況下把消息均衡的分布到主題的所有分區上，而并不關心特定消息會被寫入哪個分區。不過，生產者也可以把消息直接寫到指定的分區。這通常通過消息鍵和分區器來實現，分區器為鍵生成一個散列值，并將其映射到指定的分區上。生產者也可以自定義分區器，根據不同的業務規則將消息映射到分區。
消費者（訂閱者）讀取消息，消費者可以訂閱一個或者多個主題，并按照消息生成的順序讀取它們。消費者通過檢查消息的偏移量來區分已經讀取過的消息。偏移量是一種元數據，它是一個不斷遞增的整數值，在創建消息時，kafka會把它添加到消息里。在給定的分區里，每個消息的偏移量都是唯一的。消費者把每個分區最后讀取的消息偏移量保存在zookeeper或者kafka上，如果消費者關閉或者重啟，它的讀取狀態不會丟失。
消費者是消費者組的一部分，也就是說，會有一個或者多個消費共同讀取一個主題。消費者組保證每個分區只能被同一個組內的一個消費者使用。如果一個消費者失效，群組里的其他消費者可以接管失效消費者的工作。
1.5 broker和集群
broker：一個獨立的kafka服務器被稱為broker。broker接收來自生產者的消息，為消息設置偏移量，并提交消息到磁盤保存。broker為消費者提供服務，對讀取分區的請求作出相應，返回已經提交到磁盤上的消息。
集群：交給同一個zookeeper集群來管理的broker節點就組成了kafka的集群。
broker是集群的組成部分，每個集群都有一個broker同時充當集群控制器的角色。控制器負責管理工作，包括將分區分配給broker和監控broker。在broker中，一個分區從屬于一個broker，該broker被稱為分區的首領。一個分區可以分配給多個broker（Topic設置了多個副本的時候），這時會發生分區復制。如下圖：
broker如何處理請求：broker會在它所監聽的每個端口上運行一個Acceptor線程，這個線程會創建一個連接并把它交給Processor線程去處理。Processor線程（也叫網絡線程）的數量是可配的，Processor線程負責從客戶端獲取請求信息，把它們放進請求隊列，然后從響應隊列獲取響應信息，并發送給客戶端。如下圖所示：
生產請求和獲取請求都必須發送給分區的首領副本（分區Leader）。如果broker收到一個針對特定分區的請求，而該分區的首領在另外一個broker上，那么發送請求的客戶端會收到一個“非分區首領”的錯誤響應。Kafka客戶端要自己負責把生產請求和獲取請求發送到正確的broker上。
客戶端如何知道該往哪里發送請求呢？客戶端使用了另外一種請求類型——元數據請求。這種請求包含了客戶端感興趣的主題列表，服務器的響應消息里指明了這些主題所包含的分區、每個分區都有哪些副本，以及哪個副本是首領。元數據請求可以發給任意一個broker，因為所有的broker都緩存了這些信息。客戶端緩存這些元數據，并且會定時從broker請求刷新這些信息。此外如果客戶端收到“非首領”錯誤，它會在嘗試重新發送請求之前，先刷新元數據。
1.6 Kafka 基礎架構
二、Kafka架構深入
2.1 Kafka工作流程及文件存儲機制
2.1.1 工作流程
Kafka中消息是以topic進行分類的，生產者生產消息，消費者消費消息，都是面向topic的。
Topic是邏輯上的概念，而partition（分區）是物理上的概念，每個partition對應于一個log文件，該log文件中存儲的就是producer生產的數據。Producer生產的數據會被不斷追加到該log文件末端，且每條數據都有自己的offset。消費者組中的每個消費者，都會實時記錄自己消費到哪個offset，以便出錯恢復時，從上次的位置繼續消費。
2.1.2 文件存儲機制
由于生產者生產的消息會不斷追加到log文件末尾，為防止log文件過大導致數據定位效率低下，Kafka采取了分片和索引的機制，將每個partition分為多個segment。（由log.segment.bytes決定，控制每個segment的大小，也可通過log.segment.ms控制，指定多長時間后日志片段會被關閉）每個segment對應兩個文件——“.index”文件和“.log”文件。這些文件位于一個文件夾下，該文件夾的命名規則為：topic名稱+分區序號。例如：bing這個topic有3個分區，則其對應的文件夾為：bing-0、bing-1和bing-2。
索引文件和日志文件命名規則：每個 LogSegment 都有一個基準偏移量，用來表示當前 LogSegment 中第一條消息的 offset。偏移量是一個 64位的長整形數，固定是20位數字，長度未達到，用 0 進行填補。如下圖所示：
index和log文件以當前segment的第一條消息的offset命名。index文件記錄的是數據文件的offset和對應的物理位置，正是有了這個index文件，才能對任一數據寫入和查看擁有O(1)的復雜度，index文件的粒度可以通過參數log.index.interval.bytes來控制，默認是是每過4096字節記錄一條index。下圖為index文件和log文件的結構示意圖：
查找message的流程（比如要查找offset為170417的message）：
- 首先用二分查找確定它是在哪個Segment文件中，其中0000000000000000000.index為最開始的文件，第二個文件為0000000000000170410.index（起始偏移為170410+1 = 170411），而第三個文件為0000000000000239430.index（起始偏移為239430+1 = 239431）。所以這個offset = 170417就落在第二個文件中。其他后續文件可以依此類推，以起始偏移量命名并排列這些文件，然后根據二分查找法就可以快速定位到具體文件位置。
- 用該offset減去索引文件的編號，即170417 - 170410 = 7，也用二分查找法找到索引文件中等于或者小于7的最大的那個編號。可以看出我們能夠找到[4，476]這組數據，476即offset=170410 + 4 = 170414的消息在log文件中的偏移量。
- 打開數據文件（0000000000000170410.log），從位置為476的那個地方開始順序掃描直到找到offset為170417的那條Message。
2.1.3 數據過期機制
當日志片段大小達到log.segment.bytes指定的上限（默認是1GB）或者日志片段打開時長達到log.segment.ms時，當前日志片段就會被關閉，一個新的日志片段被打開。如果一個日志片段被關閉，就開始等待過期。當前正在寫入的片段叫做活躍片段，活躍片段永遠不會被刪除，所以如果你要保留數據1天，但是片段包含5天的數據，那么這些數據就會被保留5天，因為片段被關閉之前，這些數據無法被刪除。
2.2 Kafka生產者
2.2.1 分區策略
為什么要分區
- 多Partition分布式存儲，利于集群數據的均衡。
- 并發讀寫，加快讀寫速度。
- 加快數據恢復的速率：當某臺機器掛了，每個Topic僅需恢復一部分的數據，多機器并發。
分區的原則
- 指明partition的情況下，使用指定的partition；
- 沒有指明partition，但是有key的情況下，將key的hash值與topic的partition數進行取余得到partition值；
- 既沒有指定partition，也沒有key的情況下，第一次調用時隨機生成一個整數（后面每次調用在這個整數上自增），將這個值與topic可用的partition數取余得到partition值，也就是常說的round-robin算法。
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) {
//key為空時，獲取一個自增的計數，然后對分區做取模得到分區編號
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
// key不為空時，通過key的hash對分區取模（疑問：為什么這里不像上面那樣，使用availablePartitions呢？）
// 根據《Kafka權威指南》Page45理解：為了保證相同的鍵，總是能路由到固定的分區，如果使用可用分區，那么因為分區數變化，會導致相同的key，路由到不同分區
// 所以如果要使用key來映射分區，最好在創建主題的時候就把分區規劃好
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
private int nextValue(String topic) {
//為每個topic維護了一個AtomicInteger對象，每次獲取時+1
AtomicInteger counter = topicCounterMap.get(topic);
if (null == counter) {
counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
AtomicInteger currentCounter = topicCounterMap.putIfAbsent(topic, counter);
if (currentCounter != null) {
counter = currentCounter;
}
}
return counter.getAndIncrement();
}
2.2.2 數據可靠性保證
kafka提供了哪些方面的保證
- kafka可以保證分區消息的順序。如果使用同一個生產者往同一個分區寫入消息，而且消息B在消息A之后寫入，那么kafka可以保證消息B的偏移量比消息A的偏移量大，而且消費者會先讀取到消息A再讀取消息B。
- 只有當消息被寫入分區的所有副本時，它才被認為是“已提交”的。生產者可以選擇接收不同類型的確認，比如在消息被完全提交時的確認、在消息被寫入分區首領時的確認，或者在消息被發送到網絡時的確認。
- 只要還有一個副本是活躍的，那么已經提交的信息就不會丟失。
- 消費者只能讀取到已經提交的消息。
復制
Kafka的復制機制和分區的多副本架構是kafka可靠性保證的核心。把消息寫入多個副本可以使kafka在發生奔潰時仍能保證消息的持久性。
kafka的topic被分成多個分區，分區是基本的數據塊。每個分區可以有多個副本，其中一個是首領。所有事件都是發給首領副本，或者直接從首領副本讀取事件。其他副本只需要與首領副本保持同步，并及時復制最新的事件。
Leader維護了一個動態的in-sync replica set（ISR），意為和leader保持同步的follower集合。當ISR中的follower完成數據同步后，leader就會發送ack。如果follower長時間未向leader同步數據，則該follower將被踢出ISR，該時間閾值由replica.lag.time.max.ms參數設定。Leader不可用時，將會從ISR中選舉新的leader。滿足以下條件才能被認為是同步的：
- 與zookeeper之間有一個活躍的會話，也就是說，它在過去的6s（可配置）內向zookeeper發送過心跳。
- 在過去的10s（可配置）內從首領那里獲取過最新的數據。
影響Kafka消息存儲可靠性的配置
ack應答機制
對于某些不太重要的數據，對數據的可靠性要求不是很高，能夠容忍數據的少量丟失，所以沒有必要等ISR中的follower全部接收成功。所以Kafka提供了三種可靠性級別，用戶可以根據對可靠性和延遲的要求進行權衡。acks：
- 0： producer不等待broker的ack，這一操作提供了一個最低的延遲，broker一接收到還沒寫入磁盤就已經返回，當broker故障時可能丟失數據；
- 1： producer等待leader的ack，partition的leader落盤成功后返回ack，如果在follower同步成功之前leader故障，那么將會丟失數據；
- -1（all）：producer等待broker的ack，partition的leader和ISR里的follower全部落盤成功后才返回ack。但是如果在follower同步完成后，broker發送ack之前，leader發生故障，那么會造成重復數據。（極端情況下也有可能丟數據：ISR中只有一個Leader時，相當于1的情況）。
消費一致性保證
（1）follower故障
follower發生故障后會被臨時踢出ISR，待該follower恢復后，follower會讀取本地磁盤記錄的上次的HW，并將log文件高于HW的部分截取掉，從HW開始向leader進行同步。
等該follower的LEO大于等于該Partition的HW，即follower追上leader之后，就可以重新加入ISR了。
（2）leader故障
leader發生故障后，會從ISR中選出一個新的leader，之后為了保證多個副本之間的數據一致性，其余的follower會先將各自的log文件高于HW的部分截掉，然后從新的leader同步數據。
注意：這只能保證副本之間的數據一致性，并不能保證數據不丟失或者不重復。
2.2.3 消息發送流程
Kafka 的producer 發送消息采用的是異步發送的方式。在消息發送過程中，涉及到了兩個線程——main線程和sender線程，以及一個線程共享變量——RecordAccumulator。main線程將消息發送給RecordAccumulator，sender線程不斷從RecordAccumulator中拉取消息發送到Kafka broker。
為了提高效率，消息被分批次寫入kafka。批次就是一組消息，這些消息屬于同一個主題和分區。（如果每一個消息都單獨穿行于網絡，會導致大量的網絡開銷，把消息分成批次傳輸可以減少網絡開銷。不過要在時間延遲和吞吐量之間做出權衡：批次越大，單位時間內處理的消息就越多，單個消息的傳輸時間就越長）。批次數據會被壓縮，這樣可以提升數據的傳輸和存儲能力，但要做更多的計算處理。
相關參數：
- batch.size：只有數據積累到batch.size后，sender才會發送數據。（單位：字節，注意：不是消息個數）。
- linger.ms：如果數據遲遲未達到batch.size，sender等待 linger.ms之后也會發送數據。（單位：毫秒）。
- client.id：該參數可以是任意字符串，服務器會用它來識別消息的來源，還可用用在日志和配額指標里。
- max.in.flight.requests.per.connection：該參數指定了生產者在收到服務器響應之前可以發送多少個消息。它的值越高，就會占用越多的內存，不過也會提升吞吐量。把它設置為1可以保證消息時按發送的順序寫入服務器的，即使發生了重試。
2.3 Kafka消費者
2.3.1 消費方式
consumer采用pull（拉）的模式從broker中讀取數據。
push（推）模式很難適應消費速率不同的消費者，因為消息發送速率是由broker決定的。它的目標是盡可能以最快的速度傳遞消息，但是這樣容易造成consumer來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。而pull模式可以根據consumer的消費能力以適當的速率消費消息。
pull模式的不足之處是，如果kafka沒有數據，消費者可能會陷入循環中，一直返回空數據。針對這一點，kafka的消費者在消費數據時會傳入一個時長參數timeout，如果當前沒有數據可消費，consumer會等待一段時間后再返回。
2.3.2 分區分配策略
一個consumer group中有多個consumer，一個topic有多個partition，所以必然會涉及到partition的分配問題，即確定哪個partition由哪個consumer來消費。Kafka提供了3種消費者分區分配策略：RangeAssigor、RoundRobinAssignor、StickyAssignor。
PartitionAssignor接口用于用戶定義實現分區分配算法，以實現Consumer之間的分區分配。消費組的成員訂閱它們感興趣的Topic并將這種訂閱關系傳遞給作為訂閱組協調者的Broker。協調者選擇其中的一個消費者來執行這個消費組的分區分配并將分配結果轉發給消費組內所有的消費者。Kafka默認采用RangeAssignor的分配算法。
2.3.2.1 RangeAssignor
RangeAssignor對每個Topic進行獨立的分區分配。對于每一個Topic，首先對分區按照分區ID進行排序，然后訂閱這個Topic的消費組的消費者再進行排序，之后盡量均衡的將分區分配給消費者。這里只能是盡量均衡，因為分區數可能無法被消費者數量整除，那么有一些消費者就會多分配到一些分區。分配示意圖如下：
分區分配的算法如下：
@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
Map<String, Subscription> subscriptions) {
Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
Map<String, List<TopicPartition>> assignment = new HashMap<>();
for (String memberId : subscriptions.keySet())
assignment.put(memberId, new ArrayList<TopicPartition>());
//for循環對訂閱的多個topic分別進行處理
for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
String topic = topicEntry.getKey();
List<String> consumersForTopic = topicEntry.getValue();
Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
if (numPartitionsForTopic == null)
continue;
//對消費者進行排序
Collections.sort(consumersForTopic);
//計算平均每個消費者分配的分區數
int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
//計算平均分配后多出的分區數
int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
//計算第i個消費者，分配分區的起始位置
int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
//計算第i個消費者，分配到的分區數量
int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
}
}
return assignment;
}
這種分配方式明顯的一個問題是隨著消費者訂閱的Topic的數量的增加，不均衡的問題會越來越嚴重，比如上圖中4個分區3個消費者的場景，C0會多分配一個分區。如果此時再訂閱一個分區數為4的Topic，那么C0又會比C1、C2多分配一個分區，這樣C0總共就比C1、C2多分配兩個分區了，而且隨著Topic的增加，這個情況會越來越嚴重。分配結果：
訂閱2個Topic，每個Topic4個分區，共3個Consumer
- C0：[T0P0，T0P1，T1P0，T1P1]
- C1：[T0P2，T1P2]
- C2：[T0P3，T1P3]
2.3.2.2 RoundRobinAssignor
RoundRobinAssignor的分配策略是將消費組內訂閱的所有Topic的分區及所有消費者進行排序后盡量均衡的分配（RangeAssignor是針對單個Topic的分區進行排序分配的）。如果消費組內，消費者訂閱的Topic列表是相同的（每個消費者都訂閱了相同的Topic），那么分配結果是盡量均衡的（消費者之間分配到的分區數的差值不會超過1）。如果訂閱的Topic列表是不同的，那么分配結果是不保證“盡量均衡”的，因為某些消費者不參與一些Topic的分配。
以上兩個topic的情況，相比于之前RangeAssignor的分配策略，可以使分區分配的更均衡。不過考慮這種情況，假設有三個消費者分別為C0、C1、C2，有3個Topic T0、T1、T2，分別擁有1、2、3個分區，并且C0訂閱T0，C1訂閱T0和T1，C2訂閱T0、T1、T2，那么RoundRobinAssignor的分配結果如下：
看上去分配已經盡量的保證均衡了，不過可以發現C2承擔了4個分區的消費而C1訂閱了T1，是不是把T1P1交給C1消費能更加的均衡呢？
2.3.2.3 StickyAssignor
StickyAssignor分區分配算法，目的是在執行一次新的分配時，能在上一次分配的結果的基礎上，盡量少的調整分區分配的變動，節省因分區分配變化帶來的開銷。Sticky是“粘性的”，可以理解為分配結果是帶“粘性的”——每一次分配變更相對上一次分配做最少的變動。其目標有兩點：
- 分區的分配盡量的均衡。
- 每一次重分配的結果盡量與上一次分配結果保持一致。
當這兩個目標發生沖突時，優先保證第一個目標。第一個目標是每個分配算法都盡量嘗試去完成的，而第二個目標才真正體現出StickyAssignor特性的。
StickyAssignor算法比較復雜，下面舉例來說明分配的效果（對比RoundRobinAssignor），前提條件：
- 有4個Topic：T0、T1、T2、T3，每個Topic有2個分區。
- 有3個Consumer：C0、C1、C2，所有Consumer都訂閱了這4個分區。
上面紅色的箭頭代表的是有變動的分區分配，可以看出，StickyAssignor的分配策略，變動較小。
2.3.3 offset的維護
由于Consumer在消費過程中可能會出現斷電宕機等故障，Consumer恢復后，需要從故障前的位置繼續消費，所以Consumer需要實時記錄自己消費到哪個位置，以便故障恢復后繼續消費。Kafka0.9版本之前，Consumer默認將offset保存在zookeeper中，從0.9版本開始，Consumer默認將offset保存在Kafka一個內置的名字叫_consumeroffsets的topic中。默認是無法讀取的，可以通過設置consumer.properties中的exclude.internal.topics=false來讀取。
2.3.4 kafka高效讀寫數據（了解）
順序寫磁盤
Kafka 的 producer生產數據，要寫入到log文件中，寫的過程是一直追加到文件末端，為順序寫。數據表明，同樣的磁盤，順序寫能到600M/s，而隨機寫只有100K/s。這與磁盤的機械結構有關，順序寫之所以快，是因為其省去了大量磁頭尋址的時間。
零拷貝技術
零拷貝主要的任務就是避免CPU將數據從一塊存儲拷貝到另外一塊存儲，主要就是利用各種零拷貝技術，避免讓CPU做大量的數據拷貝任務，減少不必要的拷貝，或者讓別的組件來做這一類簡單的數據傳輸任務，讓CPU解脫出來專注于別的任務。這樣就可以讓系統資源的利用更加有效。
