在kafka中，生产者发送的消息最终会落在主题下的某个分区，但是很多开发者在使用的过程中其实并没有指定消息发往哪个分区，那么kafka是如何处理的呢？
在kafka中，消息主要由两部分组成，一部分是key，消息的键，一部分是value，消息的载体，也是实际要处理的消息内容。其中key的作用就是起到路由的作用，决定了value发往哪个分区，但前提是生产者消息对象没有明确指定消息发往哪个分区，key的路由才会发挥作用，而key的路由最终就是分区器处理的。
本文就以基于Java的kafka客户端，通过分析下源码，来了解下分区器的原理和使用
首先，通过kafka的shell命令创建一个名称为testTopic，分区数为3的topic
./kafka-topics.sh --create --topic testTopic --partitions 3 --replication-factor 1
--bootstrap-server localhost:9092
接着，启动一个消费者对主题为testTopic，partition为0的分区进行消息的监听
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --partition 0
然后，用Java客户端代码进行消息的发送，下面我们省略了大部分代码，只关注消息的构建和消息的发送
ProducerRecord<String, String> record = new ProducerRecord<>("myTopic", 0,null,"hello world");
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
producer.send(record);
在上述这段代码中，ProducerRecord就是表示消息对象，构造参数有4个值，分别为主题、分区、key、value，在这段代码中，明确指定了分区为0，因此在消费者端会有hello world的输出
consumer.png
当将ProducerRecord对象的分区数设置为1，再发送，这时发现消费者端看不到新发送的消息对象，因为明确指定了发往的分区是1，而消费者监听的分区为0，因此收不到。
因此，这边可以得出一个结论，若消息对象显示指定了消息所属的分区，那么消息就会往这个分区发送
接着，对消息对象进行下修改，不指定分区，设置一个key值，然后来跟踪下源码，看下消息会最终发往哪个分区
ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "home","hello world");
通过跟进send方法，会看到如下源码
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
省略....
Cluster cluster = clusterAndWaitTime.cluster;
byte[] serializedKey;
try {
serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
} catch (ClassCastException cce) {
throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +
" to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
" specified in key.serializer", cce);
}
byte[] serializedValue;
try {
serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
} catch (ClassCastException cce) {
throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +
" to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
" specified in value.serializer", cce);
}
int partition = partition(record, serializedKey, serializedValue, cluster);
tp = new TopicPartition(record.topic(), partition);
省略...
} catch (Exception e) {
省略...
}
}
这段代码中，这边列举出关键的部分，首先获取到集群的信息Cluster对象，然后对消息的key和value序列化成字节数组，再调用partition方法，此方法便是获取消息发送的分区序号
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
Integer partition = record.partition();
return partition != null ?
partition :
partitioner.partition(
record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}
这段代码就可以很明确的看出，首先会拿到消息对象的partition属性值，若该值不为空，则直接返回；当值为空的时候才会调用分区器的分区方法，换句话说，分区器只有在消息对象的partition属性值为null的情况下才起作用，而本文的重点就是关注分区器的作用，在kafka中分区器是一个接口，需要具体的实现，而kakfa提供一个默认的分区器实现可以直接使用，无需开发者自己手动实现，当然可以根据实际的业务需求，来实现具体的分区器
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (keyBytes == null) {
return stickyPartitionCache.partition(topic, cluster);
}
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
默认的分区器中，首先会根据是key的字节数组是否为空，若不为空，先获取集群中该主题的总分区数量，然后再根据字节数组数的一个32位hash值和总分区数进行取模运算，得到最终的分区序号，该方式也比较简单，比较复杂的是当key为null的情况，处理如下
public int partition(String topic, Cluster cluster) {
Integer part = indexCache.get(topic);
if (part == null) {
return nextPartition(topic, cluster, -1);
}
return part;
}
public int nextPartition(String topic, Cluster cluster, int prevPartition) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
Integer oldPart = indexCache.get(topic);
Integer newPart = oldPart;
// Check that the current sticky partition for the topic is either not set or that the partition that
// triggered the new batch matches the sticky partition that needs to be changed.
if (oldPart == null || oldPart == prevPartition) {
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() < 1) {
Integer random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
newPart = random % partitions.size();
} else if (availablePartitions.size() == 1) {
newPart = availablePartitions.get(0).partition();
} else {
while (newPart == null || newPart.equals(oldPart)) {
Integer random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
newPart = availablePartitions.get(random % availablePartitions.size()).partition();
}
}
// Only change the sticky partition if it is null or prevPartition matches the current sticky partition.
if (oldPart == null) {
indexCache.putIfAbsent(topic, newPart);
} else {
indexCache.replace(topic, prevPartition, newPart);
}
return indexCache.get(topic);
}
return indexCache.get(topic);
}
首先，当消息没指定key时，kafka会尽量使同一批次的消息都发往同一个分区，称之为粘性分区（sticky partition），而且会将粘性分区缓存在一个map中，这样下次可以直接从缓存取而无需重新计算分区号。
当从缓存map中查不到主题对应的粘性分区号时，调用nextPartition方法获取一个新的分区号，并将其放入到缓存中。
总结，分区器的主要作用是在消息对象没有明确指定partition值时起的作用，并且会根据key是否有值来计算。若有，则使用取模运算，因此，key值相同的消息会都发往相同的分区中。若没有，则会尽量时发往同一主题的消息都落在同一分区上，当然了，这里需要结合后续的累加器作用，才能更好的理解。
网友评论