初一下电脑课书做网站,镇江seo快速排名,广州冼村房价,微信运营包括哪些内容Kafka 系列第二篇#xff0c;详解分区机制原理。为了不错过更新#xff0c;请大家将本号“设为星标”。 点击上方“后端开发技术”#xff0c;选择“设为星标” #xff0c;优质资源及时送达 上一篇文章介绍了 Kafka 的基本概念和术语#xff0c;里面有个概念是 分区(Part… Kafka 系列第二篇详解分区机制原理。为了不错过更新请大家将本号“设为星标”。 点击上方“后端开发技术”选择“设为星标” 优质资源及时送达 上一篇文章介绍了 Kafka 的基本概念和术语里面有个概念是 分区(Partition)。 kafka 将 一个Topic 中的消息分成多份分别存储在不同的 Broker 里这每一段消息被 kafka 称为分区其中每条消息只会保存在一个分区中。 如果不太理解请回顾上一篇 开始学习 Kafka一文掌握基本概念Kafka 系列 一 为什么有分区 为什么要有分区呢 Kafka 的分区机制的本质就是将一个大的 Topic 进行拆分将一组很大的队列拆分成了多组队列。这样做有以下几个好处 因为一个 Topic 中的消息可能非常多多到一台Broker存不下因此需要拆分成多段存储在不同的机器里实现负载均衡。 拆分成多个队列可以在多个生产者和消费者的情况下发挥多机性能可以分流和并行处理消息从而提高读写性能提升系统的吞吐力。 有利于系统扩缩容提高系统的可扩展性。不同分区在不同的broker上可以通过增加新机器提高吞吐并且增加新机器的时候可以通过调整分区的分布来调配负载。 但是分区数不是越多越好需要根据系统具体情况来设置。比如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策略内容如下 如果在消息中指定了分区优先使用指定的分区。 如果没有指定分区但存在分区键则根据序列化key使用murmur2哈希算法对分区数取模。 如果没有指定分区或分区键则会使用粘性分区策略。关于粘性分区策略后面讲解 在实际生产中我们一般都默认使用此策略无需修改。 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 partitionreturn 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 粘性分区策略 实现类为 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中 然后多条消息批量发送。这样可以减少网络请求次数提高消息的发送效率。 所以批量发送消息有两个条件 一个batch满了与 batch.size有关一般大小是16k。 linger.ms时间到了。 满足任意一个条件都会触发sender线程的发送。如果生产的消息较少batch没有满就必须等到等待时间到了这就导致了较长的延迟。 因为ProducerBatch是多个为了让消息尽可能快的发送就需要让其中一个ProducerBatch先变满。 private final ConcurrentMapTopicPartition, DequeProducerBatch batches; 注意一个分区对应一个双端队列DequeProducerBatch。 粘性分区策略就是在相同的分区中优先填满一个ProducerBatch发送再去填充另一个ProducerBatch。参见下图第一个分区会被优先塞满并发送。 在一个 ProducerBatch 发送结束选择新分区的时候是随机选择的之后便会继续优先填满新的分区。 可用分区1 所有分区中随机选择。 可用分区1选择这个分区。 可用分区1所有可用分区中随机选择。 public int nextPartition(String topic, Cluster cluster, int prevPartition) {ListPartitionInfo 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) {ListPartitionInfo 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以此类推。 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {ListPartitionInfo partitions cluster.partitionsForTopic(topic);// 分区数int numPartitions partitions.size();// 下一个自增值int nextValue nextValue(topic);// 获取此主题的可用分区列表ListPartitionInfo 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 partitionreturn Utils.toPositive(nextValue) % numPartitions;}
} hash 键的值并不会影响到数据的分布这应该是数据均匀度最好的策略可以保证消息最大程度的平均分配到所有分区。 除了官方提供的策略我们还可以实现自己的分区策略比如随机策略实现起来也很简单比如按照业务键去分区的策略比如按照ip分区的策略等。 最后欢迎大家提问和交流。 加入讨论群是升职加薪第一步 回复加群 点赞是一种美德如对您有帮助欢迎评论和分享感谢阅读 实战总结记一次消息队列堆积的问题排查 2023-07-18 从二叉查找树到B*树一文搞懂搜索树的演进原创 2023-05-23 CAP、BASE理论真的很重要分布式事务系列一 2023-05-06
