第 9 章 处理大量消息

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

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

您还可以将相同的配置选项用于 Kafka Connect 运行时源连接器（包括 MirrorMaker 2.0）和接收器连接器使用的生产者和使用者。

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

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

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

使用 linger.ms 添加等待时间（以毫秒为单位），以在制作者负载下降时延迟生成请求。延迟意味着如果更多记录在最大批处理大小下，可以将更多记录添加到批处理中。

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

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

对于 MirrorMaker 2.0，目标 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

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