问题描述
我正在尝试使用 Kafka 的 RoundRobinPartitioner 类在所有分区之间均匀分布消息。我的Kafka主题配置如下：
名称：multischemakafkatopicodd
分区数：16
复制因子：2
比如说，如果我生成 100 条消息，那么每个分区应该有 6 或 7 条消息。但是，我得到了类似的东西：
sh /usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetoffsetShell --broker-list 10.0.55.211:9092 --topic multischemakafkatopicodd --time -1
multischemakafkatopicodd:0:26
multischemakafkatopicodd:5:0
multischemakafkatopicodd:10:24
multischemakafkatopicodd:15:0
multischemakafkatopicodd:13:0
multischemakafkatopicodd:8:26
multischemakafkatopicodd:2:26
multischemakafkatopicodd:12:24
multischemakafkatopicodd:14:24
multischemakafkatopicodd:9:0
multischemakafkatopicodd:11:0
multischemakafkatopicodd:4:26
multischemakafkatopicodd:1:0
multischemakafkatopicodd:6:24
multischemakafkatopicodd:7:0
multischemakafkatopicodd:3:0
我想可能是我没有产生足够的消息，所以我尝试使用 1M 记录并将分区数设置为奇数：
主题：multischemakafkatopicodd
分区数：31
复制因子：2
...我明白了。这次每个分区的消息数量分布有些均匀。
sh /usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetoffsetShell --broker-list 10.0.55.211:9092 --topic multischemakafkatopicodd --time -1
multischemakafkatopicodd:0:33845
multischemakafkatopicodd:5:34388
multischemakafkatopicodd:10:33837
multischemakafkatopicodd:20:33819
multischemakafkatopicodd:15:33890
multischemakafkatopicodd:25:34414
multischemakafkatopicodd:30:33862
multischemakafkatopicodd:26:34066
multischemakafkatopicodd:9:34088
multischemakafkatopicodd:11:34124
multischemakafkatopicodd:16:33802
multischemakafkatopicodd:4:34061
multischemakafkatopicodd:17:34977
multischemakafkatopicodd:3:34084
multischemakafkatopicodd:24:33849
multischemakafkatopicodd:23:34111
multischemakafkatopicodd:13:34062
multischemakafkatopicodd:28:33876
multischemakafkatopicodd:18:34098
multischemakafkatopicodd:22:34058
multischemakafkatopicodd:8:34079
multischemakafkatopicodd:2:33839
multischemakafkatopicodd:12:34075
multischemakafkatopicodd:29:34132
multischemakafkatopicodd:19:33924
multischemakafkatopicodd:14:34109
multischemakafkatopicodd:1:34088
multischemakafkatopicodd:6:33832
multischemakafkatopicodd:7:34080
multischemakafkatopicodd:27:34188
multischemakafkatopicodd:21:34684
我再次进行了相同的测试，但将分区数量减少到 8，我得到了这个结果，我们可以清楚地看到一些分区有接近 15K 的消息，而其他分区有大约 10K：
multischemakafkatopicodd:0:155927
multischemakafkatopicodd:5:105351
multischemakafkatopicodd:1:107382
multischemakafkatopicodd:4:160533
multischemakafkatopicodd:6:158007
multischemakafkatopicodd:7:105608
multischemakafkatopicodd:2:157934
multischemakafkatopicodd:3:105599
我做错了什么还是它应该如何工作？ 为什么消息分布如此不平等？
如果有人能帮助我，那就太好了。谢谢。
解决方法
据我所知，分区器运行良好。但您必须了解生产者所做的优化以最大限度地提高性能：
-
生产者不会为每次发送调用将每条消息生成到不同的分区，因为这会矫枉过正。
-
Round-Robin保证了类似的分布，但可以批量发送。这意味着，它会根据 {{1} }的代码：RoundRobinPartitioner
int part = Utils.toPositive(nextValue) % availablePartitions.size();
是一个 nextValue，每次分区/发送调用都会增加 1。因此，余数将始终增加 1 （以循环方式，例如有 4 个分区：AtomicInteger），假设在此过程中没有分区被声明为不可用。如果发生这种情况，循环可能看起来像 0-1-2-3-0-1-2-3-...
示例
- 主题有 4 个分区
- 每个分区的生产者分区线程缓冲区保存3条消息
（消息号计数器以 0 - 0-1-2-(partition4fails)-0-1-2-(partition4OK)-3-0-... 开头）
new AtomicInteger(0)当生成第 9 条消息时，第一个分区的缓冲区已满（因为它已经保存了 3 条消息），因此准备好发送到 kafka。如果您立即停止该过程，4 个分区将如下所示：
MsgN % Partitions Partition
0%4 0
1%4 1
2%4 2
3%4 3
4%4 0
5%4 1
6%4 2
7%4 3
8%4 0
... ...
在生成第 10 条消息时，第二个分区的缓冲区也将准备好发送出去，主题将如下所示：
Partition Offset
0 3
1 0
2 0
3 0
在现实生活中，缓冲区通常会保存大量消息（这也可以调整）。例如，假设存储了 1000 条消息。对于相同的场景，分区看起来像：
Partition Offset
0 3
1 3
2 0
3 0
因此增加了分区之间的“视觉”差异。更大的批量大小/缓冲区大小会更加臭名昭著。
这与生产者的 Partition Offset
0 1000
1 1000
2 0
3 0
线程本身的性质有关：默认情况下，它不会独立发送每条消息，而是将它们存储以便在每次代理调用时remainder，优化系统性能。
批处理是提高效率的重要驱动因素之一，并且能够实现 批处理 Kafka 生产者将尝试在内存中积累数据 并在单个请求中发送更大的批次
如果生产者停止/启动，这种不平衡可能会更加臭名昭著，因为无论先前选择的分区如何，它都会重新启动机制（因此它可以开始发送到在停止之前选择的同一分区，因此增加了与上次执行的其他非选举分区的差异).
在新的执行中，缓冲区将全部为空，因此无论哪个分区接收最多，进程都会重新启动。
因此，您在此处停止该过程：
partitioner保存每个主题的消息数量计数器的映射重新启动，因为它不是代理的一部分，来自生产者的 send multiple messages 类。如果生产者没有正确关闭和/或刷新，那些缓存的消息也将丢失。所以，在这种情况下，你得到的是之前逻辑的重复：
Partition Offset
0 1000
1 1000
2 0
3 0
这会在某些时候导致这种情况：
MsgN % Partitions Partition
0%4 0
1%4 1
2%4 2
3%4 3
(...)
这是由发送过程的非连续执行产生的不平衡，但它超出了 Partition Offset
0 2000
1 2000
2 0
3 0
的界限，其性质基于连续过程（不间断）。
您可以通过在发送消息时检查每个分区的偏移量来验证此行为：只有当所选分区存储 n 消息时，下一个选举分区将获得其批量 n 消息。
注意：示例中显示的数字引用了“完美”场景；在现实生活中，无论缓冲区大小如何，消息也可能被撤销、压缩、失败、刷新，分区不可用，......导致偏移量，如您的问题所示。
最后一个冲洗场景示例：
RoundRobinPartitioner进程已停止，但生产者已正确关闭并刷新其消息，因此主题如下所示：
Partition Offset
0 1000
1 1000
2 0
3 0
进程重新启动。刷新第一个分区的缓冲区后，看起来像：
Partition Offset
0 1997
1 1996
2 999
3 998
因此增加了对该机制的“公平性”的混淆。但这不是它的错，因为在分区器的映射、计数器和缓冲区中没有持久性。如果您让流程连续执行数天，您会发现它确实以“near-equal”的方式平衡了消息。
but of the Partitioner 的相关方法：
Partition Offset
0 2997
1 1996
2 999
3 998