多处理 kafka 消息答案

【问题标题】：multiprocessing kafka messages多处理 kafka 消息
【发布时间】：2019-06-21 05:51:39
【问题描述】：

我们有一个 kafka 消费者，它不断地轮询 kafka 主题并处理消息。我们面临的问题是我们应该为max.poll.records 赋予多少价值。为了处理每条消息，我们有 3 个休息电话。我正在做的是使用并行流将消息作为列表和处理。实例有 2GB 内存空间，max.poll.records 是 8。所以，现在如果我增加max.poll.records，那么消息处理时间会增加，当max.poll.records 为 4 时效果最好。您能否说明一下我们如何使用 1 个使用 1 个分区的实例来增加消息的并行处理，这意味着而不是获取 8 条消息，我还能尝试使其处理大约 5 条消息。

更新：- 下面是内存配置，我在pcf中部署独立应用。

JVM内存配置：

-Xmx1422566K -Xss1M -XX:ReservedCodeCacheSize=240M -XX:MaxDirectMemorySize=10M -XX:MaxMetaspaceSize=162585K

【问题讨论】：

你的问题不是很清楚。您想要实现什么？并行处理是什么意思？在 Kafka 中，通过分区和多个消费者实例来提高并行度。
我的意思是说，我想从 kafka 获取大约 20 条消息并并行处理它们。假设我有一个消费者，只有一个分区。所以，我想增加 max.poll.records 并且不影响整体消息处理时间。

标签： java multithreading apache-kafka kafka-consumer-api

【解决方案1】：

max.poll.records 不影响并行度。根据documentation，

max.poll.records: 单次返回的最大记录数调用 poll()。

如果您增加此配置的值，您将能够在每次轮询时获取更多记录，但这并不意味着您将并行处理消息。

Kafka 的并行度由分区数量和启动并运行的消费者实例的数量控制。

一个 Kafka 主题被划分为多个分区，这些分区通过在多个代理之间拆分数据来实现并行性。多个分区允许多个消费者并行读取一个主题。因此，为了实现并行处理，您需要将您的主题划分为多个分区。

为了增加现有主题的分区数量，您可以简单地运行

bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic topicName --partitions 10

不过，这不会移动现有数据

关于消费者、消费者组和分区的说明
如果你有 N 个分区，那么在同一个消费者组中最多可以有 N 个消费者，每个消费者都从一个分区中读取。当您的消费者少于分区时，一些消费者将从多个分区中读取。此外，如果您的消费者多于分区，那么一些消费者将处于非活动状态，并且根本不会从这些实例中收到任何消息。

【讨论】：

谢谢，知道了。你能帮我解决问题吗：-我有 5000 条主题消息，我正在获取 8 条记录并处理它们，因此处理总共 5k 条记录大约需要 63 秒。我想获取更多记录以及如何减少处理时间。那么，有什么策略可以解决这个问题。这样我还可以减少对经纪人的投票。
@RahulSingh 您需要做的是增加主题的分区数量（例如将其设置为 3），然后启动 3 个属于 same 消费者组。
听起来不错，但我期待在 java 端优化一些方法，轮询更多记录并处理它们。有什么想法吗？
@RahulSingh 如果您想在每次投票中处理更多记录，那么只需增加 max.poll.records。
是的，我做到了，但是它的性能受到影响，然后处理 5k 条消息需要更多时间，不知道为什么会出现这种行为。

【解决方案2】：

作为对Giorgos回答的补充，此澄清来自Kafka the definitive guide

在选择数量时需要考虑几个因素分区：

您希望为该主题实现的吞吐量是多少？例如，您希望每秒写入 100 KB 还是 1 GB？

从单个分区消费时，您期望达到的最大吞吐量是多少？您将始终拥有最多一个消费者从分区读取，所以如果你知道你的消费者速度较慢将数据写入数据库，该数据库永远不会处理更多每个线程写入它的速度超过每秒 50 MB，那么你知道从分区消费时，您的吞吐量限制为 60MB。

您可以通过相同的练习来估计单个分区的每个生产者的最大吞吐量，但由于生产者通常比消费者快得多，跳过通常是安全的这个。

如果您根据键向分区发送消息，稍后添加分区可能非常具有挑战性，因此请计算吞吐量基于您预期的未来使用情况，而不是当前使用情况。

考虑您将在每个代理上放置的分区数量以及每个代理的可用磁盘空间和网络带宽。

避免高估，因为每个分区都使用了broker上的内存和其他资源，并且会增加leader的时间选举。

考虑到所有这些，很明显您需要多个分区，但不是很多。

如果您对目标有一些估计主题的吞吐量和消费者的预期吞吐量，你可以划分目标吞吐量由预期的消费者吞吐量得出，并以此方式得出分区数。所以如果我希望能够从主题中写入和读取 1 GB/秒，我知道每个消费者只能处理 50 MB/s，那么我知道我至少需要 20 分区。这样，我可以让 20 个消费者阅读该主题并达到 1 GB/秒。如果您没有这些详细信息，我们的经验表明，限制分区的大小每天保留少于 6 GB 的磁盘通常会令人满意结果。

【讨论】：