Partition Strategies kafka分区计谋

原文链接 Kafka Partition Strategy
分区计谋简介
分区增长了并行化并允许Kafka扩展。存在许多将消息分发到主题分区的计谋。在我们深入研究每种计谋的配景之前，下面的表格给出了每种计谋的扼要概述。
Kafka消息由生产者发送，由斲丧者吸取。这两种计谋是差别的，以是我们有两个表，一个总结每种计谋。
生产者分区计谋

计谋	形貌
Default partitioner	key=null，利用 round-robin 方式。 key!=null, 利用 hash 算法
Round-robin partitioner	round-robin 方式
Uniform sticky partitioner	消息被发送到一个粘性分区(直到 batch.size 符合 或 linger.ms time 到时间 )以镌汰耽误。
Custom partitioner	这种方法实现了Partitioner接口，用一些界说key-to-partition路由计谋的自界说逻辑覆盖 partition 方法。

Default partitioner

顾名思义，这是天生消息的默认计谋。当键为空时，纪录被随机发送到主题的一个可用分区。如果一个键存在，Kafka将对该键举行散列，并利用效果将消息映射到特定的分区。这可确保具有雷同键的消息终极位于雷同分区中。但是，只有在主题中的分区数目保持稳固的情况下，这种映射才是同等的：如果添加了新分区，那么具有雷同键的新消息大概会被写入与具有雷同键的旧消息差别的分区。
【下图左边是按时间排序（chronological）的message， 从右到左，右边K1:Va 第一个发送】
【下图右边是根据默认分区器路由到3个partition的效果， 添加次序是从右到左。 雷同key的message被分配到同一个partition(但同一个分区会出现差别的key)， 将按发生次序举行斲丧。】

Round robin partitioner

当生产者盼望在全部分区之间匀称分配写利用时，利用这种方法。这种分布与键的哈希值（或键为空）无关，因此具有雷同键的消息终极大概位于差别的分区中。
【下图左边是按时间排序（chronological）的message， 从右到左，右边K1:Va 第一个发送】

当工作负载因单个键而发生毛病时，即为同一键天生许多消息时，此计谋非常有用。假设消息的次序无关紧急，而且利用默认分区器。在这种情况下，负载不平衡将导致消息在分区中列队，并增长分配给这些分区的斲丧者子集的负载。轮询计谋将导致消息在各个分区之间匀称分布。

Uniform sticky partitioner

现在，当没有指定分区和键时，生产者的默认分区步调以轮询方式对纪录举行分区。这意味着一系列一连纪录中的每条纪录将被发送到差别的分区，直到覆盖全部分区，然后生产者重新开始。固然这会在分区之间匀称地分布纪录，但也会产生更多较小的批，从而导致更多的哀求和列队以及更高的耽误。
为了办理这一题目，引入了匀称粘分器。它有两条规则：
1，如果用纪录指定了分区，则按原样利用该分区。
2，如果没有指定分区，则会选择一个sticky分区，直到设置文件里的属性 “batch.size” 被填满或 “ linger.ms ” （发送消息前等候的时间）到了。

“固定 （Sticking）”到一个分区可以实现更大的 batch，并镌汰体系中的耽误。发送 batch 后，sticky分区发生变革。随着时间的推移，纪录匀称地分布在全部分区中。
纪录键不作为分区计谋的一部分利用，因此不能包管具有雷同键的纪录被发送到雷同的分区。
【下图左边是按时间排序（chronological）的message， 从右到左，右边K1:Va 第一个发送】
【下图右边是利用 Uniform Sticky 分区计谋的效果， batch满了发送，3个消息一个 batch的发送。】

Custom partitioner

偶然，用例与任何标准分区器都不太得当。比方，假设我们想要将事件日志数据写入Kafka，此中一个用户（称为“CEO”）占全部事件的40%以上。
如果利用默认散列分区，“CEO”用户的纪录将被分配到与其他用户雷同的分区。这将导致一个分区比其他分区大得多，从而导致署理耗尽空间并减慢处置惩罚速率。抱负的办理方案是为用户“CEO”提供一个专用分区，然后利用散列分区将别的用户映射到剩余的分区。下面是这个用例的简单实现：

1. import org.apache.kafka.clients.producer.Partitioner;
2. import org.apache.kafka.common.Cluster;
3. import org.apache.kafka.common.PartitionInfo;
4. import org.apache.kafka.common.record.InvalidRecordException;
5. import org.apache.kafka.common.utils.Utils;
7. public class PowerUserPartitioner implements Partitioner {
9. public void configure(Map<String, ?> configs) {}
11. public int partition(String topic, Object key, byte[] keyBytes,
12. Object value, byte[] valueBytes,Cluster cluster) {
14. List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
15. int numPartitions = partitions.size();
16. if ((keyBytes == null) || (!(key instanceOf String)))
17.       throw new InvalidRecordException("Record must have a valid string key");
19. if (((String) key).equals("CEO"))
20.       return numPartitions - 1; // Messages with key “CEO” will always go to the last partition
22. // Other records will get hashed to the rest of the partitions
23. return Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1);
24. }
26. public void close() {}
28. }

复制代码

斲丧者分配计谋

计谋	形貌
Range assignor (default)	（Partitions 总数） / （Consumers总数）个 partitions被分配到每个consumer。 其目的是实现共处分区，即将两个差别主题的雷同分区号分配给雷同的斲丧者。（P0 of Topic X and P0 of Topic Y 到同一个consumer）
Round-robin assignor	分区被单独挑选并分配给斲丧者（按照任何公道的次序，比如从前到后）。当全部的斲丧者都用完，但仍有一些分区未分配时，将从第一个斲丧者开始重新分配。如许做的目的是最大化利用的斲丧者数目。
Sticky assignor	这种方法的工作原理类似于轮循分配，但在重新分配分区时生存尽大概多的现有分配。目的是镌汰或完全克制再平衡过程中的分区移动。
Custom assignor	继续 AbstractPartitionAssignor 然后重写  assign method

斲丧者和斲丧者组（Consumers and consumer groups）

斲丧者是从Kafka读取数据的应用步调。通常，我们有多个生产者向主题写入消息，因此从主题读取和处置惩罚数据的单个斲丧者大概无法跟上传入消息的速率，而且越来越掉队。为了从主题扩展斲丧，Kafka有一个叫做斲丧者组的概念。顾名思义，斲丧者群体就是一群斲丧者。当斲丧者组中的多个斲丧者订阅同一主题时，每个斲丧者吸取来自主题中差别分区集的消息，从而在它们之间分发数据。
唯一须要注意的是，一个分区只能分配给单个斲丧者。向组中添加更多的斲丧者有助于扩大斲丧规模。但是，如果斲丧者的数目多于分区，则一些斲丧者将保持空闲状态，由于没有任何分区可供它们利用。这个空闲的斲丧者充当故障转移斲丧者，允许它在现有斲丧者发生故障时敏捷增补空缺。

斲丧者分区分配

就像我们可以控制怎样将数据写入主题中的分区一样，Kafka也允许我们决定斲丧者怎样从这些分区中读取数据。我们可以设置用于在斲丧者实例之间分配分区的计谋。属性 “partition.assignment.strategy” 可用于在设置斲丧者时设置分配计谋。

【上图展示了一个topic有三个 partition 怎样分配给四个斲丧者组】
【Consumer Group 1 拥有一个斲丧者 C11， 负责全部的 Partitions】
【Consumer Group 2 拥有两个斲丧者 C21和C22， C21负责全部的 Partition 0， P1，P2由C22负责】
【Consumer Group 3 拥有三个斲丧者， 各自负责一个Partition】
【Consumer Group 4 拥有四个斲丧者， 各自负责一个Partition， 空闲状态的C44时间准备接替挂掉的斲丧者】
以下是斲丧者分区分配的一些计谋：
范围分配器（Range assignor）：

这是默认计谋，并在每个主题（ per-topic）的根本上工作。对于每个主题，可用分区按数字次序思量，斲丧者按字典次序（）思量。然后将分区数除以斲丧者数，以确定分配给每个斲丧者的分区数。如果它没有匀称分别，那么前几个斲丧者将有一个额外的分区。
轮循分配器（Round robin assignor）：

它担当全部分区，并以轮循方式将它们分配给斲丧者。这里的上风在于，它旨在最大化利用的斲丧者数目。但是，由于这个缘故起因，在重新平衡的情况下，它不能镌汰分区移动。
Sticky分配器（Sticky assignor）：

它的工作原理与轮询分配器非常相似，但有一个额外的长处，由于它在发生分区重新分配时生存了尽大概多的现有分配。当主题分区在重新平衡期间从一个斲丧者移动到另一个斲丧者时，这有助于节省一些开销处置惩罚（将在背面的末节中讨论）。
自界说分配器（Custom assignor）：

也可以为分配器编写自界说逻辑。下面是一个基于分配给斲丧者的优先级处置惩罚故障转移的示例

1. public class CustomPartitionAssignor extends AbstractPartitionAssignor implements Configurable {
3.   @Override
4.   public void configure(final Map<String, ?> configMap) {
5.     // define the configs here
6.   }
8.   @Override
9.   public String name() {
10.     return "CustomPartitionerAssigner";
11.   }
13. @Override
14.   public Subscription subscription(final Set<String> topicSet) {
15.     return new Subscription(
16.         new ArrayList<>(topicSet),
17.         ByteBuffer.wrap(ByteBuffer.allocate(4).putInt(config.priority()))
18.      );
19.   }
21.   @Override
22.   Map<String, List<TopicPartition>> assign(
23.                          Map<String, Integer> partitionsTopicMap,
24.                          Map<String, Subscription> subscriptionsMap)  {
26.     Stream<ConsumerPriority> consumerOrdered = subscriptionsMap.entrySet()
27.         .stream()
28.         .map(x -> {
29.             int consumerPriority = x.getValue().data().getInt();
30.             String consumerId = x.getKey();
31.             return new ConsumerPriority(consumerId, consumerPriority);
32.         })
33.         .sorted(Comparator.reverseOrder());
35.     ConsumerPriority priority = consumerOrdered.findFirst().get();
37.     final List<TopicPartition> assignments = partitionsTopicMap
38.         .entrySet()
39.         .stream()
40.         .flatMap(entry -> {
41.             final String topicName = entry.getKey();
42.             final int partitionsCount = entry.getValue();
43.             return IntStream.range(0, partitionsCount).mapToObj( i -> new TopicPartition(topicName, i));
44.         }).collect(Collectors.toList());
46.     final Map<String, List<TopicPartition>> assignmentMap = new HashMap<>();
47.     subscriptionsMap.keySet().forEach(memberId -> assignmentMap.put(memberId, Collections.emptyList()));
48.     assignmentMap.put(priority.memberId, assignments);
49.     return assignmentMap;
50.   }
51. }

复制代码

这个自界说分配器可以在初始化斲丧者时利用。

1. Properties properties = new Properties();
2. ...
3. ...
4. properties.put(
5.     ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,   
6.     CustomPartitionAssignor.class.getName()
7. );
8. ...
9. ...
10. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

复制代码

斲丧者的再平衡（Consumer rebalancing）


斲丧者可以随时添加到斲丧者组或从斲丧者组中删除。当添加新的斲丧者时，它开始从先前分配给差别斲丧者的分区中斲丧消息。类似地，当一个斲丧者被删除时，它的分区被分配给另一个剩余的活动斲丧者。这种重新分配和将分区全部权从一个斲丧者转移到另一个斲丧者的过程称为再平衡。重新平衡是至关紧张的，由于它们提供了高可用性和可伸缩性，允许用户安全地添加和删除斲丧者。
有两种范例的再平衡。利用哪一个取决于斲丧者利用的分区分配计谋：
急于再平衡：（Eager rebalancing: ）

全部斲丧者克制斲丧，放弃其分区的全部权，重新到场组，然后得到分配给它们的新分区。当整个组中没有斲丧者时，这会导致一个小的停机窗口（斲丧者不可用性）。
协作再平衡：（Cooperative rebalancing: ）

也称为增量再平衡（ incremental rebalancing），此计谋在多个阶段实行再平衡。它涉及到将一小部分分区从一个斲丧者重新分配给另一个斲丧者，允许斲丧者继续处置惩罚来自未重新分配的分区的消息，并克制完全不可用。

在正常情况下，重新平衡大概是不可取的。在重新平衡期间，斲丧者在一段时间内克制处置惩罚消息，这会导致处置惩罚来自主题的变乱的耽误。对于须要实时变乱处置惩罚而且不能蒙受高出几秒钟的耽误的用例来说，这大概是有题目的。
Kafka提供了一种风趣的方式来克制这种重新平衡。
静态构成员（Static group membership）

默认的斲丧者再平衡举动是组中的斲丧者身份是临时的。当斲丧者离开组时，它的分区被取消；当它重新到场时，它会得到一个新的成员ID，并为它分配一组新的分区。
通过利用唯一的“group.instance”设置斲丧者，可以使其成为静态构成员。id”属性。当它到场一个组时，将根据所利用的分区分配计谋为它分配一组分区。但是，它在重新启动或关闭时不会主动离开组——它仍旧是一个成员。在重新毗连时，它被辨以为其唯一的静态标识，并被重新分配到它所利用的雷同分区，而不会触发再平衡。
这对于有状态应用步调非常有用，此中状态由分配给斲丧者的分区添补。由于这个过程大概很耗时，因此每次斲丧者重启时都重新创建这个初始状态或缓存是不抱负的。

结论

我们已经看到，从生产者和斲丧者的角度来看，找到准确的分区计谋是多么紧张。如果没有准确设置，从久远来看，这大概会成为一个题目。决定很大水平上取决于用例的渺小差异、数据量等，没有一种方法能办理全部题目。
默认选项大概实用于大多数情况，但偶然它们不是准确的选择。比方，如果在生产者端不须要排序，那么循环或同一粘性计谋的性能会好得多。类似地，在斲丧者端，为用例选择默认范围分配器，正如我们在事件日志示例（倾斜负载）中讨论的那样，大概会导致不须要的再平衡和痴钝的处置惩罚。
允许自界说分区器和分配器，但它们并不简单，须要深入的技能明白才气准确实现。找到准确的计谋是得到Kafka提供的速率和可扩展性上风的关键一步。