基於Kafka客戶端的高階API，配合zookeeper的使用，可以有效的實現Kafka叢集的Rebalance，提高生產環境下的健壯性。本文使用librdkafka(https://github.com/edenhill/librdkafka) 提供的高階API來實現生產/消費，作為測試的基礎。

生產者的負載均衡
對於同一個Topic的不同Partition，Kafka會盡力將這些Partition分佈到不同的Broker伺服器上，這種均衡策略實際上是基於Zookeeper實現的。在一個Broker啟動時，會首先完成Broker的註冊過程，並註冊一些諸如“有哪些可訂閱的Topic”之類的元資料資訊。生產者啟動後也要到zookeeper下注冊，建立一個臨時節點來監聽Broker伺服器列表的變化。由於在Zookeeper下Broker建立的也是臨時節點，當Brokers發生變化時，生成者可以得到相關的通知，從改變自己的Broker list。其他的諸如Topic的變化以及Broker和Topic的關係變化，也是通過Zookeeper的這種Watcher監聽實現的。
在生產中，必須指定topic；但是對於partition,有兩種指定方式：

1. 明確指定partition(0-N)，則資料被髮送到指定partition
2. 設定為RD_KAFKA_PARTITION_UA，則kafka會回撥partitioner進行均衡選取，partitioner方法需要自己實現。可以輪詢或者傳入key進行hash。未實現則採用預設的隨機方法rd_kafka_msg_partitioner_random隨機選擇。

消費者的負載均衡
Kafka具有消費分組的概念，某個Topic的某個partition只能由一個Consumer group中的一個Consmer消費。但如果兩個Consmer不在同一個Consumer group，那麼他們是可以同時消費某Topic的同一個partition的。
對於某些低級別的API，Consumer消費時必須制定topic和partition，這顯然不是一種很好的均衡策略。基於高級別的API，Consumer消費時只需制定topic，藉助zookeeper可以根據partition的數量和consumer的數量做到均衡的動態配置。
消費者在啟動時會到zookeeper下以自己的conusmer-id建立臨時節點/consumer/[group-id]/ids/[conusmer-id]，並對/consumer/[group-id]/ids註冊監聽事件，當消費者發生變化時，同一group的其餘消費者會得到通知。當然，消費者還要監聽broker列表的變化。librdkafka通常會將partition進行排序後，根據消費者列表，進行輪流的分配。消費者負載均衡中涉及的更重要的一點還有如何動態的維護partition的offset，因為消費者可能是變化的，而offset需要由comsumer來維護，librdkafka是藉助zookeeper和消費者佇列實現的，具體可以檢視librdkafka的實現原始碼。(ps:

Kafka已推薦將consumer的位移資訊儲存在Kafka內部的topic中，即__consumer_offsets(/brokers/topics/__consumer_offsets)，並且預設提供了kafka_consumer_groups.sh指令碼供使用者檢視consumer資訊(sh kafka-consumer-groups.sh –bootstrap-server * –describe –group *)。在當前版本中，offset儲存方式要麼儲存在本地檔案中，要麼儲存在broker端, 具體的儲存方式取決”offset.store.method”的配置，預設是儲存在broker端)

Rebalance測試
下面基於librdKafka實現的C/C++客戶端（0.9版本及以上），
測試Kafka叢集的負載均衡。
測試環境及方法：

• 三個broker通過zookeeper組成的叢集；
• 每個Topic預設三個分割槽，每個分割槽預設只有一個副本；
• 通過增減生產者、消費者、broker來觀察生產/消費的變化

1.配置kafka叢集
由於條件有限，在同一個機器上啟動三個broker來模擬kafka叢集，三個broker使用另外安裝的同一個zookeeper服務（實際叢集中，每個broker通常在不同的機器上，也會使用不同host的zookeeper）
zookeeper的安裝和啟動並不複雜，在此略過。

cd /home/kafka/config
//準備三份用於啟動kafka服務的配置
cp server.properties server-0.properties
cp server.properties server-1.properties
cp server.properties server-2.properties
三份配置中都要修改以下
broker.id=0(三個配置中分別修改為0,1,2)
port=9092(三個配置中分別修改為9092,9093,9094)
log.dirs=/tmp/kafka-logs-0(三個配置中分別修改為/tmp/kafka-logs-0,/tmp/kafka-logs-1,/tmp/kafka-logs-2)
num.partitions=3 (都設定為3，即每個topic預設三個partition)

2.測試
生產消費的負載均衡
啟動三個broker：修改完配置後，開啟三個shell視窗，分別啟動三個broker
kafka/bin/kafka-server-start.sh server-X.properties
啟動日誌

...
[2017-02-19 20:38:39,766] INFO Kafka version : 0.10.1.1 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:38:39,766] INFO Kafka commitId : f10ef2720b03b247 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:38:39,775] INFO [Kafka Server 0], started (kafka.server.KafkaServer)

...
[2017-02-19 20:39:12,628] INFO Kafka version : 0.10.1.1 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:39:12,629] INFO Kafka commitId : f10ef2720b03b247 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:39:12,632] INFO [Kafka Server 1], started (kafka.server.KafkaServer)

...
[2017-02-19 20:42:18,809] INFO Kafka version : 0.10.1.1 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:42:18,809] INFO Kafka commitId : f10ef2720b03b247 (org.apache.kafka.common.utils.AppInfoParser)
[2017-02-19 20:42:18,812] INFO [Kafka Server 2], started (kafka.server.KafkaServer)

啟動兩個producer，建立名為xyz的topic，根據配置，該topic預設有3個partition:

[root@localhost producer]# ./producer 
Current RdKafka-ver:0.9.4-pre1
% Created producer rdkafka#producer-1
kafka的broker的日誌變化分別是
//broker-0，名為xyz的topic的partition-2被分別到broker-0
[2017-02-19 20:45:17,268] INFO Partition [xyz,2] on broker 0: No checkpointed highwatermark is found for partition [xyz,2] (kafka.cluster.Partition)
//broker-1，名為xyz的topic的partition-0被分別到broker-1
[2017-02-19 20:45:17,325] INFO Partition [xyz,0] on broker 1: No checkpointed highwatermark is found for partition [xyz,0] (kafka.cluster.Partition)
broker-2，名為xyz的topic的partition-1被分別到broker-2
[2017-02-19 20:45:17,001] INFO Partition [xyz,1] on broker 2: No checkpointed highwatermark is found for partition [xyz,1] (kafka.cluster.Partition)
在/tmp/kafka-logs-[*]的變化也可以印證這一點

依次啟動三個consumer：

//啟動1臺時，該消費者負責消費所有的3個partition
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[0], xyz[1], xyz[2],

//啟動2臺時，會觸發消費者的負載均衡
//consumer-1
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[0], xyz[1], xyz[2], 
RebalanceCb: Local: Revoke partitions: xyz[0], xyz[1], xyz[2], 
RebalanceCb: Local: Assign partitions: xyz[0], xyz[1],

//consumer-2
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[2],

//啟動3臺時，會再次觸發消費者的負載均衡
//consumer-1
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[0], xyz[1], xyz[2], 
RebalanceCb: Local: Revoke partitions: xyz[0], xyz[1], xyz[2], 
RebalanceCb: Local: Assign partitions: xyz[0], xyz[1], 
RebalanceCb: Local: Revoke partitions: xyz[0], xyz[1], 
RebalanceCb: Local: Assign partitions: xyz[0],

//consumer-2
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[2], 
RebalanceCb: Local: Revoke partitions: xyz[2], 
RebalanceCb: Local: Assign partitions: xyz[1],

//consumer-3
[[email protected] consumer]# ./consumer 
Current RdKafka-ver:0.9.4-pre1
RebalanceCb: Local: Assign partitions: xyz[2],

當依次關閉consumer時，同樣會觸發類似的rebalance，再次不一一演示
使用兩個producer輪流向topic-xyz傳送訊息，三個消費者會大致消費相同的訊息數目

broker的均衡
如果此時關掉一個broker，那麼所以的消費者和生產者客戶端都會收到如下的提示

2017-02-19 21:06:02.206: LOG-3-FAIL: [thrd:localhost:9094/bootstrap]: localhost:9094/bootstrap: Connect to ipv4#127.0.0.1:9094 failed: Connection refused
2017-02-19 21:06:02.206: ERROR (Local: Broker transport failure): localhost:9094/bootstrap: Connect to ipv4#127.0.0.1:9094 failed: Connection refused
在此之後，producer傳送的訊息將不會被髮送host為localhost:9094的broker；如果該broker又恢復了，那麼producer和consumer都會在此連線上該服務，因為他們都聽過zookeeper監聽著broker的變化

PS:
a.在本文的情況下（一個topic只有三個partition），如果啟動4個consumer了？
由於只有3個partition，那麼最後會有一個consumer無法消費

b.Rebalance後的消費者的從哪兒開始消費了？
offset由客戶端和zookeeper通過consumer佇列維護全域性的變化