0


详解Kafka分区机制原理|Kafka 系列 二

Kafka 系列第二篇,详解分区机制原理。为了不错过更新,请大家将本号“设为星标”。

点击上方“后端开发技术”,选择“设为星标” ,优质资源及时送达

上一篇文章介绍了 Kafka 的基本概念和术语,里面有个概念是 分区(Partition)。

kafka 将 一个Topic 中的消息分成多份,分别存储在不同的 Broker 里,这每一段消息被 kafka 称为分区,其中每条消息只会保存在一个分区中。

如果不太理解请回顾上一篇:

开始学习 Kafka,一文掌握基本概念|Kafka 系列 一

为什么有分区?

为什么要有分区呢?

Kafka 的分区机制的本质就是将一个大的 Topic 进行拆分,将一组很大的队列拆分成了多组队列。这样做有以下几个好处:

  1. 因为一个 Topic 中的消息可能非常多,多到一台Broker存不下,因此需要拆分成多段存储在不同的机器里,实现负载均衡。
  2. 拆分成多个队列,可以在多个生产者和消费者的情况下发挥多机性能,可以分流和并行处理消息,从而提高读写性能,提升系统的吞吐力。
  3. 有利于系统扩缩容,提高系统的可扩展性。不同分区在不同的broker上,可以通过增加新机器提高吞吐,并且增加新机器的时候可以通过调整分区的分布来调配负载。
    0de38ad040a27e5ddabd8449fac6d55e.png
    但是分区数不是越多越好,需要根据系统具体情况来设置。比如3个Broker就应该至少有3个分区,如果broker性能之间有差异,可以调大分区数进行调配。也可以通过broker的倍数来设置分区数,并且进行性能压测,测试集群的吞吐量。

分区数过多会带来资源管理上的消耗,清除日志时间变长,集群broker故障后分区leader重选时间变长,客户端消费端线程数需求增加,甚至导致连接所需的socket消耗增加。

分区策略

分区策略就是决定生产者将会把消息发送到具体哪个分区的算法,分区策略由

Partitioner

接口实现。

自定义分区策略

用于分区的

partition

方法定义如下:

/**
     * Compute the partition for the given record.
     *
     * @param topic topic名 The topic name
     * @param key 用于分区的key The key to partition on (or null if no key)
     * @param keyBytes 用于分区的序列号key 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
     */
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

可以看出,这里提供了 Topic 和一些跟消息有关的key参数,cluster 是集群信息,包含Kafka 当前的Node 数据以及Topic、partition数据等。有了这些数据,具体拿到一条消息该发往哪个分区,我们就可以根据已有信息制定自己的分区策略。

# name of the partitioner class for partitioning events; default partition spreads data randomly
#partitioner.class=

我们实现了自定义的 Partition 类之后,就可以设置

partitioner.class

为目标策略类,Producer 就会按照我们的自定义策略来对消息进行分区。

默认分区策略

Kafka 提供了默认分区策略

DefaultPartitioner

,策略内容如下:

  1. 如果在消息中指定了分区,优先使用指定的分区。
  2. 如果没有指定分区,但存在分区键,则根据序列化key使用murmur2哈希算法对分区数取模。
  3. 如果没有指定分区或分区键,则会使用粘性分区策略。(关于粘性分区策略后面讲解)

在实际生产中,我们一般都默认使用此策略,无需修改。

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    return partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());
}
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,
                         int numPartitions) {
    if (keyBytes == null) {
        return stickyPartitionCache.partition(topic, cluster);
    }
    // hash the keyBytes to choose a partition
    return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}

注意,这里指的分区键是序列化后的key,也就是变量 keyBytes,其他key、value、valueBytes 并没用到。

byte[] keyBytes = keySerializer.serialize(topic, record.headers(), record.key());
default byte[] serialize(String topic, Headers headers, T data) {
  // data 变量
    return serialize(topic, data);
}

看到 key 等序列化方法我们可以明白,key 的序列号值只受到 record.key() 的影响,所以同样的key会被固定分配到同样的partition中。(注意这里的key是指用于分区的key,而不是topic)
e16317cdfef524b9779fd048bbc2e591.png

粘性分区策略

实现类为

UniformStickyPartitioner 

,他与默认分区策略的区别是:

  • DefaultPartitionerd 默认分区策略:如果有分区键的话,会按照分区键来决定分区,这个时候并不会使用粘性分区策略。
  • UniformStickyPartitioner粘性分区策略:无论有没有分区键,都用粘性分区来分配。
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    return stickyPartitionCache.partition(topic, cluster);
}

什么是粘性分区策略?

我们需要知道,在Producer在发送消息的时候,会将消息放到一个ProducerBatch中, 然后多条消息批量发送。这样可以减少网络请求次数,提高消息的发送效率。

所以批量发送消息有两个条件:

  1. 一个batch满了,与 batch.size有关,一般大小是16k。
  2. linger.ms时间到了。

满足任意一个条件,都会触发sender线程的发送。如果生产的消息较少,batch没有满,就必须等到等待时间到了,这就导致了较长的延迟。

因为ProducerBatch是多个,为了让消息尽可能快的发送,就需要让其中一个ProducerBatch先变满。

private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

注意:一个分区对应一个双端队列

Deque<ProducerBatch>>

粘性分区策略就是在相同的分区中,优先填满一个ProducerBatch,发送,再去填充另一个ProducerBatch。参见下图,第一个分区会被优先塞满并发送。
fb4f66d75ce57e9231f4ea8ed6c56c88.png
在一个 ProducerBatch 发送结束,选择新分区的时候,是随机选择的,之后便会继续优先填满新的分区。

  • 可用分区<1 ,所有分区中随机选择。
  • 可用分区=1,选择这个分区。
  • 可用分区>1,所有可用分区中随机选择。
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)) {
                    int 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);
    }

轮询分区策略

Kafka 中提供了轮训策略的实现

RoundRobinPartitioner

。当用户希望将写操作均匀地分发到所有分区时,可以使用此分区策略。

举例,有三个分区,针对于同一个producer,第一条消息发送到partition1,第二条消息发送到partition2,第三条发送到partition3,以此类推。
7372d4c65e8afd01c61335a18caddd6b.png

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();
    // 下一个自增值
    int nextValue = nextValue(topic);
    // 获取此主题的可用分区列表
    List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
    if (!availablePartitions.isEmpty()) {
        // topic可用分区不为空,取余
        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;
    }
}

hash 键的值并不会影响到数据的分布,这应该是数据均匀度最好的策略,可以保证消息最大程度的平均分配到所有分区。

除了官方提供的策略,我们还可以实现自己的分区策略,比如随机策略,实现起来也很简单;比如按照业务键去分区的策略;比如按照ip分区的策略等。

最后,欢迎大家提问和交流。

加入讨论群是升职加薪第一步!

回复:加群

9cc714a9272b10f02503d017f1df6ee5.jpeg

点赞是一种美德,如对您有帮助,欢迎评论和分享,感谢阅读!

实战总结|记一次消息队列堆积的问题排查

2023-07-18

从二叉查找树到B*树,一文搞懂搜索树的演进!|原创

2023-05-23

CAP、BASE理论真的很重要!|分布式事务系列(一)

2023-05-06

标签: kafka 分布式

本文转载自: https://blog.csdn.net/sinat_32873711/article/details/132138110
版权归原作者 西海幼鸟 所有, 如有侵权,请联系我们删除。

“详解Kafka分区机制原理|Kafka 系列 二”的评论:

还没有评论