1.解耦：
　　允許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
2.冗余：
　　消息隊列把數據進行持久化直到它們已經被完全處理，通過這一方式規避了數據丟失風險。許多消息隊列所采用的"插入-獲取-刪除"范式中，在把一個消息從隊列中刪除之前，需要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。
3.擴展性：
　　因為消息隊列解耦了你的處理過程，所以增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。
4.靈活性 & 峰值處理能力：
　　在訪問量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量并不常見。如果為以能處理這類峰值訪問為標準來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵組件頂住突發的訪問壓力，而不會因為突發的超負荷的請求而完全崩潰。
5.可恢復性：
　　系統的一部分組件失效時，不會影響到整個系統。消息隊列降低了進程間的耦合度，所以即使一個處理消息的進程掛掉，加入隊列中的消息仍然可以在系統恢復后被處理。
6.順序保證：
　　在大多使用場景下，數據處理的順序都很重要。大部分消息隊列本來就是排序的，并且能保證數據會按照特定的順序來處理。（Kafka 保證一個 Partition 內的消息的有序性）
7.緩沖：
　　有助于控制和優化數據流經過系統的速度，解決生產消息和消費消息的處理速度不一致的情況。
8.異步通信：
　　很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。

1.producer：
　　消息生產者，發布消息到 kafka 集群的終端或服務。
2.broker：
　　kafka 集群中包含的服務器。
3.topic：
　　每條發布到 kafka 集群的消息屬于的類別，即 kafka 是面向 topic 的。
4.partition：
　　partition 是物理上的概念，每個 topic 包含一個或多個 partition。kafka 分配的單位是 partition。
5.consumer：
　　從 kafka 集群中消費消息的終端或服務。
6.Consumer group：
　　high-level consumer API 中，每個 consumer 都屬于一個 consumer group，每條消息只能被 consumer group 中的一個 Consumer 消費，但可以被多個 consumer group 消費。
7.replica：
　　partition 的副本，保障 partition 的高可用。
8.leader：
　　replica 中的一個角色， producer 和 consumer 只跟 leader 交互。
9.follower：
　　replica 中的一個角色，從 leader 中復制數據。
10.controller：
　　kafka 集群中的其中一個服務器，用來進行 leader election 以及 各種 failover。
12.zookeeper：
　　kafka 通過 zookeeper 來存儲集群的 meta 信息。

1. 指定了 patition，則直接使用；
2. 未指定 patition 但指定 key，通過對 key 的 value 進行hash 選出一個 patition
3. patition 和 key 都未指定，使用輪詢選出一個 patition。

//創建消息實例
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
     if (topic == null)
          throw new IllegalArgumentException("Topic cannot be null");
     if (timestamp != null && timestamp < 0)
          throw new IllegalArgumentException("Invalid timestamp " + timestamp);
     this.topic = topic;
     this.partition = partition;
     this.key = key;
     this.value = value;
     this.timestamp = timestamp;
}

//計算 patition，如果指定了 patition 則直接使用，否則使用 key 計算
private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
     Integer partition = record.partition();
     if (partition != null) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
          int lastPartition = partitions.size() - 1;
          if (partition < 0 || partition > lastPartition) {
               throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
          }
          return partition;
     }
     return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

// 使用 key 選取 patition
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 = counter.getAndIncrement();
          List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
          if (availablePartitions.size() > 0) {
               int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
               return availablePartitions.get(part).partition();
          } else {
               return DefaultPartitioner.toPositive(nextValue) % numPartitions;
          }
     } else {
          //對 keyBytes 進行 hash 選出一個 patition
          return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
     }
}

1. producer 先從 zookeeper 的 "/brokers/.../state" 節點找到該 partition 的 leader
2. producer 將消息發送給該 leader
3. leader 將消息寫入本地 log
4. followers 從 leader pull 消息，寫入本地 log 后 leader 發送 ACK
5. leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 發送 ACK

1. At most once 消息可能會丟，但絕不會重復傳輸
2. At least one 消息絕不會丟，但可能會重復傳輸
3. Exactly once 每條消息肯定會被傳輸一次且僅傳輸一次

1. 基于時間：log.retention.hours=168
2. 基于大小：log.retention.bytes=1073741824

1. controller 在 ZooKeeper 的 /brokers/topics 節點上注冊 watcher，當 topic 被創建，則 controller 會通過 watch 得到該 topic 的 partition/replica 分配。
2. controller從 /brokers/ids 讀取當前所有可用的 broker 列表，對于 set_p 中的每一個 partition：
	2.1 從分配給該 partition 的所有 replica（稱為AR）中任選一個可用的 broker 作為新的 leader，并將AR設置為新的 ISR
	2.2 將新的 leader 和 ISR 寫入 /brokers/topics/[topic]/partitions/[partition]/state
3. controller 通過 RPC 向相關的 broker 發送 LeaderAndISRRequest。

1. controller 在 zooKeeper 的 /brokers/topics 節點上注冊 watcher，當 topic 被刪除，則 controller 會通過 watch 得到該 topic 的 partition/replica 分配。
2. 若 delete.topic.enable=false，結束；否則 controller 注冊在 /admin/delete_topics 上的 watch 被 fire，controller 通過回調向對應的 broker 發送 StopReplicaRequest。

1. 將所有 broker（假設共 n 個 broker）和待分配的 partition 排序
2. 將第 i 個 partition 分配到第（i mod n）個 broker 上
3. 將第 i 個 partition 的第 j 個 replica 分配到第（(i + j) mode n）個 broker上

1. 等待 ISR 中的任一個 replica 活過來，并選它作為 leader。可保障數據不丟失，但時間可能相對較長。
2. 選擇第一個活過來的 replica（不一定是 ISR 成員）作為 leader。無法保障數據不丟失，但相對不可用時間較短。

1. controller 在 zookeeper 的 /brokers/ids/[brokerId] 節點注冊 Watcher，當 broker 宕機時 zookeeper 會 fire watch
2. controller 從 /brokers/ids 節點讀取可用broker
3. controller決定set_p，該集合包含宕機 broker 上的所有 partition
4. 對 set_p 中的每一個 partition
    4.1 從/brokers/topics/[topic]/partitions/[partition]/state 節點讀取 ISR
    4.2 決定新 leader（如4.3節所描述）
    4.3 將新 leader、ISR、controller_epoch 和 leader_epoch 等信息寫入 state 節點
5. 通過 RPC 向相關 broker 發送 leaderAndISRRequest 命令

1. 讀取并增加 Controller Epoch。
2. 在 reassignedPartitions Patch(/admin/reassign_partitions) 上注冊 watcher。
3. 在 preferredReplicaElection Path(/admin/preferred_replica_election) 上注冊 watcher。
4. 通過 partitionStateMachine 在 broker Topics Patch(/brokers/topics) 上注冊 watcher。
5. 若 delete.topic.enable=true（默認值是 false），則 partitionStateMachine 在 Delete Topic Patch(/admin/delete_topics) 上注冊 watcher。
6. 通過 replicaStateMachine在 Broker Ids Patch(/brokers/ids)上注冊Watch。
7. 初始化 ControllerContext 對象，設置當前所有 topic，“活”著的 broker 列表，所有 partition 的 leader 及 ISR等。
8. 啟動 replicaStateMachine 和 partitionStateMachine。
9. 將 brokerState 狀態設置為 RunningAsController。
10. 將每個 partition 的 Leadership 信息發送給所有“活”著的 broker。
11. 若 auto.leader.rebalance.enable=true（默認值是true），則啟動 partition-rebalance 線程。
12. 若 delete.topic.enable=true 且Delete Topic Patch(/admin/delete_topics)中有值，則刪除相應的Topic。

1. The high-level Consumer API
2. The SimpleConsumer API

1. 如果消費線程大于 patition 數量，則有些線程將收不到消息
2. 如果 patition 數量大于線程數，則有些線程多收到多個 patition 的消息
3. 如果一個線程消費多個 patition，則無法保證你收到的消息的順序，而一個 patition 內的消息是有序的

1. 多次讀取一個消息
2. 只消費一個 patition 中的部分消息
3. 使用事務來保證一個消息僅被消費一次

1. 必須在應用程序中跟蹤 offset，從而確定下一條應該消費哪條消息
2. 應用程序需要通過程序獲知每個 Partition 的 leader 是誰
3. 需要處理 leader 的變更

1. 查找到一個“活著”的 broker，并且找出每個 partition 的 leader
2. 找出每個 partition 的 follower
3. 定義好請求，該請求應該能描述應用程序需要哪些數據
4. fetch 數據
5. 識別 leader 的變化，并對之作出必要的響應

1.讀完消息先 commit 再處理消息。
    這種模式下，如果 consumer 在 commit 后還沒來得及處理消息就 crash 了，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息，這就對應于 At most once
2.讀完消息先處理再 commit。
    這種模式下，如果在處理完消息之后 commit 之前 consumer crash 了，下次重新開始工作時還會處理剛剛未 commit 的消息，實際上該消息已經被處理過了。這就對應于 At least once。
3.如果一定要做到 Exactly once，就需要協調 offset 和實際操作的輸出。
    精典的做法是引入兩階段提交。如果能讓 offset 和操作輸入存在同一個地方，會更簡潔和通用。這種方式可能更好，因為許多輸出系統可能不支持兩階段提交。比如，consumer 拿到數據后可能把數據放到 HDFS，如果把最新的 offset 和數據本身一起寫到 HDFS，那就可以保證數據的輸出和 offset 的更新要么都完成，要么都不完成，間接實現 Exactly once。（目前就 high-level API而言，offset 是存于Zookeeper 中的，無法存于HDFS，而SimpleConsuemr API的 offset 是由自己去維護的，可以將之存于 HDFS 中）

1. 將目標 topic 下的所有 partirtion 排序，存于PT
2. 對某 consumer group 下所有 consumer 排序，存于 CG，第 i 個consumer 記為 Ci
3. N=size(PT)/size(CG)，向上取整
4. 解除 Ci 對原來分配的 partition 的消費權（i從0開始）
5. 將第i*N到（i+1）*N-1個 partition 分配給 Ci

1.Herd effect
　　任何 broker 或者 consumer 的增減都會觸發所有的 consumer 的 rebalance
2.Split Brain
　　每個 consumer 分別單獨通過 zookeeper 判斷哪些 broker 和 consumer 宕機了，那么不同 consumer 在同一時刻從 zookeeper 看到的 view 就可能不一樣，這是由 zookeeper 的特性決定的，這就會造成不正確的 reblance 嘗試。
3. 調整結果不可控
　　所有的 consumer 都并不知道其它 consumer 的 rebalance 是否成功，這可能會導致 kafka 工作在一個不正確的狀態。

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = security_protocol://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092

　# Hostname and port the broker will advertise to producers and consumers. If not set, 
　# it uses the value for "listeners" if configured. Otherwise, it will use the value
　# returned from java.net.InetAddress.getCanonicalHostName().
　#advertised.listeners=PLAINTEXT://your.host.name:9092

1. listeners=PLAINTEXT://121.10.26.XXX:9092
2. advertised.listeners=PLAINTEXT://121.10.26.XXX:9092