说明
当 topic 有多个分区时，一个生产者将消息发到某个分区的策略，就叫分区策略，如下：
指定分区号，数据会直接发送到所指定的分区
没有指定分区号，指定了数据的key，可以通过key获取hashCode决定数据发送到哪个分区
既没有给定分区号，也没有给定key值，会采取round-robin fashion，是kafka的轮询策略
自定义分区策略
源码分析
调用流程如下图：
第一步
调用 KafkaProducer 类的下面方法，返回分区号：
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
Integer partition = record.partition();
//当有 分区号 时，就直接返回分区号
return partition != null ? partition : this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}
没有 分区号 时，就执行 this.partitioner.partition() 方法，见第二步
第二步
调用 DefaultPartitioner 类的下面方法：
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();
//如果key为空，轮询的方式得到分区号
if (keyBytes == null) {
//获取下一个数字，目的是轮询发送给各个分区
int nextValue = this.nextValue(topic);
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
//如果可用的 分区数量大于0
if (availablePartitions.size() > 0) {
//对可用分区数量取模
int part = Utils.toPositive(nextValue) % availablePartitions.size();
return ((PartitionInfo)availablePartitions.get(part)).partition();
} else {
//对总共分区数量取模
return Utils.toPositive(nextValue) % numPartitions;
}
} else {
//如果key不为空，就根据 keyBytes 算出 `hashCode`，在对分区数取模
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
nextValue()
在多线程中，以原子的方式 +1
private int nextValue(String topic) {
AtomicInteger counter = (AtomicInteger)this.topicCounterMap.get(topic);
//如果取出的数字为null，就创建一个随机数，并放入
if (null == counter) {
counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
AtomicInteger currentCounter = (AtomicInteger)this.topicCounterMap.putIfAbsent(topic, counter);
if (currentCounter != null) {
counter = currentCounter;
}
}
//不为空就+1，并返回
return counter.getAndIncrement();
}