上一章节我们详细 分析了Kafka在发送消息的时候，是如何拉取元数据的，正如我们源码里doSend()里waitOnMetadata介绍的：
//这里同步拉取元数据
ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
//这里我们已经拉取到元数据了，cluster里也有对应的值了
Cluster cluster = clusterAndWaitTime.cluster;
接下是key,value的序列化,这块很简单，我们不做过多的分析
/**
* 序列化key value
*/
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);
}
接下来就是我们这节要重点分析的是分区器，我们知道发送一个消息，从开始的拦截器过滤后，接着我们取到了消息的元数据信息，上面我们又对key,value进行了序列化，那么接下来，不就是获取消息发送的分区号，这样根据元数据信息，就可以知道要发送到哪个broker node上了，如下图：
/**
*
* 基于获取到的元数据，使用Partitioner组件获取消息对应的分区
*/
int partition = partition(record, serializedKey, serializedValue, cluster);
//把消息包装成一个topicPartition格式对象
tp = new TopicPartition(record.topic(), partition);
可以看到，方法里调用了partition这个方法：
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
//这里当我们在生产端发送数据时，如果我们设置了发送消息的分区号，那么此时partition不为null
//通常我们不会设置的，那么方法就进入partitioner.partition
Integer partition = record.partition();
return partition != null ?
partition :
partitioner.partition(
record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}
如果我们没有在生产端配置分区号，那么就进入了partitioner.partition()方法，在这里大家是不是对partitioner很眼熟，没错就是它，我们翻到我们第一章节里，当初始化KafkaProducer的时候，同时是不是也初始化了partitioner这个属性，如下面代码
this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
可以看到如果我们在生产端，没有配置分区器，就使用默认分区器DefaultPartitioner,这里也同时给出自定义分区器的配置partitioner.class =xxxx，关于自定义分区器，网上文章很多，很简单，这里就不过多的说明
进入DefaultPartitioner.partition()里，看详细代码：
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 = nextValue(topic);
/**
* 这里取得可用有效的topic的partitons列表
* 当key为空时，查找可用的partitions数量，
* 当大于0时，取%availablePartitions
* 当小于等于0时，取%numPartitions
*/
//获取我们要发磅消息对应的topic的可用分区信息
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
//一个自增的数对有效分区总数取模，来达到轮训的效果，达到负载均衡
//6 % 3 = 0
//7 % 3 = 1
//8 % 3 = 2
int part = Utils.toPositive(nextValue) % availablePartitions.size();
//分配分区号
return availablePartitions.get(part).partition();
} else {
// no partitions are available, give a non-available partition
//如果没有有效的分区，直接对总分区数取模
return Utils.toPositive(nextValue) % numPartitions;
}
} else {
// hash the keyBytes to choose a partition
/**
* 当key值不为空时，取keyBytes的hash
* key值相同时，会下面值一样，会进同一个分区里面
*/
//直接对key取hashCode值% 分区总数 = 分区编号，这样如果是同一个key,最后一定是发往同一个分区里了
//如果想要让消息发往一个分区，必须要指定对应的key
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
上面代码里注解我写的很详细了，这里就不做过多的说明，这样我们是不是就取得了发送消息的分区号了，这样就可以继续向下发送消息了。
本章节内容简单，但里面有不少我们可以学习的东西，比如做一个轮训的效果，我们就可以学习上面的代码，拿过来用就可，这可是工业级的代码，谁还能比这个代码写的更好？
网友评论