kafka生成者—分区器详解

灰太狼 2023-10-11 02:59 47阅读 0赞

我们知道，Kafka中的每个Topic一般会分配N个Partition，那么生产者(`Producer`)在将消息记录(`ProducerRecord`)发送到某个`Topic`对应的`Partition`时采用何种策略呢？Kafka中采用了分区器(`Partitioner`)来为我们进行分区路由的操作。本文将详细讨论Kafka给我们提供的分区器实现`DefaultPartitioner`，当然我们也可以实现自定义的分区器，只需要实现`Partitioner`接口。

### ProducerRecord ###

在具体分析分区器之前，我们先看一下生产者产生的消息记录的结构。生产者产生的每一个消息均用`ProducerRecord`来表示，其字段如下所示：

public class ProducerRecord<K, V> {
        //该消息需要发往的主题
        private final String topic;
        //该消息需要发往的主题中的某个分区，如果该字段有值，则分区器不起作用，直接发往指定的分区
        //如果该值为null，则利用分区器进行分区的选择 
        private final Integer partition;
        private final Headers headers;
        //如果partition字段为null，则使用分区器进行分区选择时会用到该key字段，该值可为空 
        private final K key;
        private final V value;
        private final Long timestamp;

本文我们暂时只关注`ProducerRecord`中的`topic`，`partition`，`key`这三个字段。

### Partitioner接口 ###

`Partitioner`接口中有一个最主要的方法:

/**
         * Compute the partition for the given record.
         *
         * @param topic The topic name
         * @param key The key to partition on (or null if no key)
         * @param keyBytes The serialized key to partition on( or null if no key)
         * @param value The value to partition on or null
         * @param valueBytes The serialized value to partition on or null
         * @param cluster The current cluster metadata
         */
    int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

这里重点关注一下最后一个参数`Cluster cluster`，目前只需要理解cluster代表了Kafka的元数据信息，从该cluster字段能够获取到我们需要的信息，在这里我们只关注从cluster能够根据指定topic，获取该topic所对应的分区的信息。

### DefaultPartitioner ###

我们先看一下生产者在发送消息时选择分区的逻辑，该逻辑在`KafkaProducer`类的`partition` 方法中:

/**
         * computes partition for given record.
         * if the record has partition returns the value otherwise
         * calls configured partitioner class to compute the 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);
        }

如上代码所示：首先判断`ProducerRecord`中的`partition`字段是否有值，即是否在创建消息记录的时候直接指定了分区，如果指定了分区，则直接将该消息发送到指定的分区，否则调用分区器的`partition`方法，执行分区策略。如果用户配置了分区器，则使用用户指定的分区器，否则使用默认的分区器，即`DefaultPartitioner`，下面我们看一下，该默认实现是如何进行分区选择的。

public class DefaultPartitioner implements Partitioner {
    
        private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap<>();
    
        /**
         * Compute the partition for the given record.
         *
         * @param topic The topic name
         * @param key The key to partition on (or null if no key)
         * @param keyBytes serialized key to partition on (or null if no key)
         * @param value The value to partition on or null
         * @param valueBytes serialized value to partition on or null
         * @param cluster The current cluster metadata
         */
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            /* 首先通过cluster从元数据中获取topic所有的分区信息 */
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            //拿到该topic的分区数
            int numPartitions = partitions.size();
            //如果消息记录中没有指定key
            if (keyBytes == null) {
                //则获取一个自增的值
                int nextValue = nextValue(topic);
                //通过cluster拿到所有可用的分区（可用的分区这里指的是该分区存在首领副本）
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                //如果该topic存在可用的分区
                if (availablePartitions.size() > 0) {
                    //那么将nextValue转成正数之后对可用分区数进行取余
                    int part = Utils.toPositive(nextValue) % availablePartitions.size();
                    //然后从可用分区中返回一个分区
                    return availablePartitions.get(part).partition();
                } else { // 如果不存在可用的分区
                    //那么就从所有不可用的分区中通过取余的方式返回一个不可用的分区
                    return Utils.toPositive(nextValue) % numPartitions;
                }
            } else { // 如果消息记录中指定了key
                // 则使用该key进行hash操作，然后对所有的分区数进行取余操作，这里的hash算法采用的是murmur2算法，然后再转成正数
                //toPositive方法很简单，直接将给定的参数与0X7FFFFFFF进行逻辑与操作。
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }
    
        //nextValue方法可以理解为是在消息记录中没有指定key的情况下，需要生成一个数用来代替key的hash值
        //方法就是最开始先生成一个随机数，之后在这个随机数的基础上每次请求时均进行+1的操作
        private int nextValue(String topic) {
            //每个topic都对应着一个计数
            AtomicInteger counter = topicCounterMap.get(topic);
            if (null == counter) { // 如果是第一次，该topic还没有对应的计数
                //那么先生成一个随机数
                counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
                //然后将该随机数与topic对应起来存入map中
                AtomicInteger currentCounter = topicCounterMap.putIfAbsent(topic, counter);
                if (currentCounter != null) {
                    //之后把这个随机数返回
                    counter = currentCounter;
                }
            }
            //一旦存入了随机数之后，后续的请求均在该随机数的基础上+1之后进行返回
            return counter.getAndIncrement();
        }

### 总结 ###

生产者发送消息时整个分区路由的步骤如下：

1.  判断消息中的`partition`字段是否有值，有值的话即指定了分区，直接将该消息发送到指定的分区就行。
2.  如果没有指定分区，则使用分区器进行分区路由，首先判断消息中是否指定了`key`。
3.  如果指定了`key`，则使用该`key`进行hash操作，并转为正数，然后对`topic`对应的分区数量进行取模操作并返回一个分区。
4.  如果没有指定`key`，则通过先产生随机数，之后在该数上自增的方式产生一个数，并转为正数之后进行取余操作。

上述第4点需要注意一下，如果该`topic`有可用分区，则**优先分配可用分区**，如果没有可用分区，则分配一个不可用分区。这与第3点中`key`有值的情况不同，`key`有值时，不区分可用分区和不可用分区，直接取余之后选择某个分区进行分配。