第 11 章 处理大量信息

如果您的 AMQ Streams 部署需要处理大量信息，您可以使用配置选项优化吞吐量和延迟。

Kafka 生成者和消费者配置可帮助控制对 Kafka 代理的请求大小和频率。有关配置选项的更多信息，请参阅以下内容：

您还可以将相同的配置选项与 Kafka Connect 运行时源连接器（包括 MirrorMaker 2）和接收器连接器使用的制作者和消费者一起使用。

对于消费者配置（consumer.*），您可以增加在单个获取请求中获取的数据量，以减少延迟。您可以使用 fetch.max.bytes 和 max.partition.fetch.bytes 属性增加 fetch 请求大小。您还可以使用 max.poll.records 属性设置从消费者缓冲区返回的消息数量的最大限制。

对于生成者配置(producer)，您可以增加在单个生成请求中发送的消息批处理的大小。您可以使用 batch.size 属性增加批处理大小。更大的批处理大小可减少准备好发送的未完成消息的数量，以及消息队列中积压的大小。发送到同一分区的消息将一起批处理。当达到批处理大小时，生成请求将发送到目标集群。通过增加批处理大小，生成请求会延迟，更多的消息添加到批处理中，并同时发送到代理。当您只有几个处理大量消息的主题分区时，这可以提高吞吐量。

考虑制作者为合适的生产批处理大小处理的记录的数量和大小。

使用 linger.ms 添加等待时间（以毫秒为单位），以延迟生产负载减少时生成请求。delay 表示如果记录位于最大批处理大小下，则可以向批处理添加更多记录。

对于 Kafka Connect 源连接器，到目标 Kafka 集群的数据流管道如下：

对于 Kafka Connect sink 连接器，数据流管道到目标外部数据源，如下所示：

对于 MirrorMaker 2，数据将管道镜像到目标 Kafka 集群，如下所示：

producer 会将缓冲区中的信息发送到目标 Kafka 集群中的主题。在发生这种情况时，Kafka Connect 任务继续轮询数据源，以将消息添加到源消息队列中。

源连接器的制作者缓冲区的大小使用 buffer.memory 属性设置。任务在清除缓冲区前等待指定的超时时间(offset.flush.timeout.ms)。这应该有足够的时间供代理和提交的偏移数据确认。源任务不会等待生成者在提交偏移前清空消息队列，但关机期间除外。

如果生产者无法满足源消息队列中消息的吞吐量，则缓冲区将被阻止，直到缓冲区中的空间在 max.block.ms 绑定的时间段内有空间可用。在此期间，缓冲区中仍然发送任何未确认的信息。在确认和刷新这些消息前，不会向缓冲区添加新消息。

您可以尝试以下配置更改，将未处理消息的底层源消息队列保持为可管理的大小：

根据可用的 CPU 和内存资源和 worker 节点数量，请考虑可以并行运行的任务数量。您可能需要调整配置值，直到它们有所需的效果。

# ...
producer.batch.size=327680
producer.linger.ms=100
# ...
tasks.max = 2

# ...
producer.batch.size=327680
producer.linger.ms=100
# ...
tasks.max = 2

# ...
consumer.fetch.max.bytes=52428800
consumer.max.partition.fetch.bytes=1048576
consumer.max.poll.records=500
# ...
tasks.max = 2

# ...
consumer.fetch.max.bytes=52428800
consumer.max.partition.fetch.bytes=1048576
consumer.max.poll.records=500
# ...
tasks.max = 2