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。
11.zookeeper：
　　kafka 通過 zookeeper 來儲存叢集的 meta 資訊。

//建立訊息例項
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;
     }
}