6.Kafka源码深入解析之分区器

上一章节我们详细 分析了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;
        }
    }

上面代码里注解我写的很详细了，这里就不做过多的说明，这样我们是不是就取得了发送消息的分区号了，这样就可以继续向下发送消息了。
本章节内容简单，但里面有不少我们可以学习的东西，比如做一个轮训的效果，我们就可以学习上面的代码，拿过来用就可，这可是工业级的代码，谁还能比这个代码写的更好？