第 9 章 处理大量消息

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

Kafka producer 和使用者配置可帮助控制对 Kafka 代理的请求的大小和频率。有关配置选项的详情请参考以下内容：

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

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

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

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

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

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

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

对于 MirrorMaker 2.0，数据镜像管道到目标 Kafka 集群如下：

制作者在其缓冲中发送消息发送到目标 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