Kafka 配置调整

如何指定配置属性取决于部署的类型。如果在 OCP 上部署 AMQ Streams，您可以使用 Kafka 资源通过 config 属性为 Kafka 代理添加配置。在 RHEL 上使用 AMQ Streams，您可以将配置作为环境变量添加到属性文件中。

当您向自定义资源添加 config 属性时，您可以使用冒号(':')来映射属性和值。

num.partitions:1

当您将属性添加为环境变量时，您可以使用等号('=')来映射属性和值。

num.partitions=1

以下工具有助于 Kafka 调整：

有关这些工具的更多信息，请参阅以下指南：

典型的代理配置将包括与主题、线程和日志相关的属性的设置。

# ...
num.partitions=1
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
num.network.threads=3
num.io.threads=8
num.recovery.threads.per.data.dir=1
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
group.initial.rebalance.delay.ms=0
zookeeper.connection.timeout.ms=6000
# ...

基本主题属性为主题设置默认分区数和复制因素，这些主题将应用于没有显式设置这些属性创建的主题，包括何时自动创建主题。

# ...
num.partitions=1
auto.create.topics.enable=false
default.replication.factor=3
min.insync.replicas=2
replica.fetch.max.bytes=1048576
# ...

对于高可用性环境，建议将复制因素增加到至少 3 个用于主题，并将最小 in-sync 副本数量设置为复制因素小 1。

auto.create.topics.enable 属性默认为启用，以便在生产者和消费者需要时自动创建不存在的主题。如果使用自动主题创建，您可以使用 num.partitions 为主题设置默认分区数量。但是，此属性会被禁用，以便通过显式主题创建为主题提供更多控制。

为了实现数据持久性，需要在 topic 配置中设置 min.insync.replicas，并在 producer 配置中使用 acks=all 来发送提交确认。

使用 replica.fetch.max.bytes 设置复制领导分区的每个后续程序获取的最大大小（以字节为单位）。根据平均消息大小和吞吐量更改这个值。在考虑读/写缓冲区所需的总内存分配时，可用内存也必须能适应所有后续者的最大复制消息大小。

delete.topic.enable 属性默认为启用，以允许删除主题。在生产环境中，您应该禁用此属性以避免意外删除主题，从而导致数据丢失。但是，您可以临时启用它并删除主题，然后再次禁用它。

# ...
auto.create.topics.enable=false
delete.topic.enable=true
# ...

如果您使用事务 启用对生成者中的分区的原子写入，则事务的状态存储在内部 __transaction_state 主题。默认情况下，代理配置有 3 个复制因素，以及此主题的最小同步副本，这意味着 Kafka 集群中最少需要三个代理。

# ...
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
# ...

同样，存储消费者状态的内部 __consumer_offsets 主题具有分区和复制因素的数量的默认设置。

# ...
offsets.topic.num.partitions=50
offsets.topic.replication.factor=3
# ...

不要在生产环境中减少这些设置。在生产环境中，您可以提高设置。作为例外，您可能需要减少单代理 测试环境中的设置。

网络线程处理对 Kafka 集群的请求，如从客户端应用程序生成和获取请求。生成请求将置于请求队列中。响应放置在响应队列中。

每个监听器的网络线程数量应该反映了复制因素以及客户端制作者和与 Kafka 集群交互的用户的活动级别。如果您要拥有大量请求，您可以使用闲置时间线程数量来增加线程数量，以确定何时添加更多线程。

要减少拥塞并规范请求流量，您可以限制请求队列中允许的请求数。当请求队列已满时，所有传入流量都会被阻断。

I/O 线程从请求队列获取请求来处理它们。添加更多线程可以提高吞吐量，但 CPU 内核和磁盘带宽的数量会实施实际的上限。至少，I/O 线程的数量应等于存储卷的数量。

# ...
num.network.threads=3 

queued.max.requests=500 

num.io.threads=8 

num.recovery.threads.per.data.dir=4 

# ...

所有代理的线程池的配置更新可能会在集群级别动态发生。这些更新仅限于当前大小的一半和两倍的当前大小。

如果线程因为磁盘数量而较慢或限制，您可以尝试增加网络请求的缓冲区的大小，以提高吞吐量：

# ...
replica.socket.receive.buffer.bytes=65536
# ...

另外，增加 Kafka 可以接收的最大字节数：

# ...
socket.request.max.bytes=104857600
# ...

Kafka 批处理数据，以便在从 Kafka 到客户端（如数据中心之间的连接）上实现合理的吞吐量。但是，如果高延迟问题，您可以增加缓冲区的大小来发送和接收信息。

# ...
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
# ...

您可以使用 bandwidth-delay 产品 计算来估算缓冲区的最佳大小，这会乘以带往返延迟（以秒为单位）链路的最大带宽（以秒为单位）以达到最大吞吐量。

Kafka 依赖于日志来存储消息数据。日志由一系列片段组成，每个片段都与基于偏移和基于时间戳的索引相关联。新消息被写入 活跃 片段，永远不会被修改。当服务从消费者获取请求时，将读取片段。活跃的片段 会定期 变为只读，并创建一个新的活跃网段来替换它。每个代理的每个主题分区只能有一个活跃网段。保留旧的片段，直到它们有资格删除。

在代理级别配置决定了日志片段的最大大小（以字节为单位），以及活跃的片段前的时间（毫秒）：

# ...
log.segment.bytes=1073741824
log.roll.ms=604800000
# ...

这些设置可以使用 segment.bytes 和 segment.ms 在主题级别上覆盖。降低或引发这些值的选择取决于删除网段的策略。较大的大小表示活跃片段包含更多信息，且会更频繁地推出。片段还有资格更频繁地删除。

在 Kafka 中，日志清理策略决定如何管理日志数据。在大多数情况下，您不需要更改集群级别的默认配置，它指定了 删除 清理策略，并启用 紧凑 清理策略使用的日志清理：

# ...
log.cleanup.policy=delete
log.cleaner.enable=true
# ...

当使用日志保留策略时，当达到保留限制时，非活跃日志片段会被删除。删除旧片段有助于防止超过磁盘容量。

对于基于时间的日志保留，您可以根据小时、分钟或毫秒设置保留周期：

# ...
log.retention.ms=1680000
# ...

保留周期基于消息附加到段中的时间。Kafka 使用片段中最新消息的时间戳来确定该片段是否已过期。毫秒配置的优先级超过分钟，其优先级高于小时。默认情况下，分钟和毫秒配置是 null，但三个选项提供了对您要保留数据的大量控制级别。首选项应提供给毫秒配置，因为它是唯一可以动态更新的三个属性之一。

如果 log.retention.ms 设置为 -1，则不会将时间限制应用到日志保留，并且保留所有日志。但是，通常不建议此设置，因为它可能会导致出现难以重新排序的完整磁盘的问题。

对于基于大小的日志保留，您可以指定最小日志大小（以字节为单位）：

# ...
log.retention.bytes=1073741824
# ...

这意味着 Kafka 将确保至少有指定数量的日志数据可用。

例如，如果您将 log.retention.bytes 设置为 1000，并将 log.segment.bytes 设为 300，则 Kafka 将保留 4 个片段加活跃片段，确保至少有 1000 字节可用。当活跃片段已满并创建新片段时，会删除最旧的网段。此时，磁盘的大小可能会超过指定的 1000 字节，可能范围介于 1200 到 1500 字节（不包括索引文件）。

使用日志大小的潜在问题是，它不会考虑时间信息被附加到段中。您可以对清理策略使用基于时间和大小的日志保留，以获得您需要的平衡。首先达到哪个阈值会触发清理。

要在从系统中删除分段文件前添加时间延迟，您可以在所有主题的代理级别使用 log.segment.delete.delay.ms ：

# ...
log.segment.delete.delay.ms=60000
# ...

或在主题级别配置 file.delete.delay.ms。

您可以设置检查日志进行清理的频率（以毫秒为单位）：

# ...
log.retention.check.interval.ms=300000
# ...

调整与日志保留设置相关的日志保留检查间隔。较小的保留大小可能需要更频繁地检查。清理的频率通常足以管理磁盘空间，但通常不会影响代理的性能。

如果为日志压缩设置了清理策略，则日志 头 作为标准 Kafka 日志运行，并按顺序写入新消息。在紧凑的日志的 尾部 中，日志清理程序运行，如果稍后日志中发生具有相同键的另一个记录，则会删除记录。具有 null 值的消息也会被删除。要使用压缩，您必须有识别相关消息的密钥，因为 Kafka 保证保留每个密钥的最新信息，但它不能保证整个紧凑的日志不会包含重复。

图 3.1. 在压缩前显示带有偏移位置的键值写入的日志

使用键识别信息，Kafka 压缩会保留日志尾部针对特定消息键中存在的最新消息（具有最高偏移），最终丢弃具有相同键的早期消息。其最新状态的消息始终可用，在日志清理程序运行时最终会删除该特定消息的任何过时的记录。您可以将消息恢复到以前的状态。即使周围记录被删除，记录也会保留其原始偏移量。因此，tail 可以具有非连续偏移。当消耗在 tail 中可用偏移时，会发现带有下一个偏移的记录。

图 3.2. 压缩后日志

如果适当，您可以在压缩过程中添加一个延迟：

# ...
log.cleaner.delete.retention.ms=86400000
# ...

删除的数据保留周期提供了在数据被不可避免删除前注意到数据的时间。

要删除与特定密钥相关的所有消息，制作者可以发送 tombstone 消息。tombstone 有一个 null 值，并作为标记来通知消费者删除该键的对应消息。一段时间后，只会保留 tombstone 标记。假设新消息继续进入，标记会保留在 log.cleaner.delete.retention.ms 指定的持续时间内，以便消费者有足够的时间识别删除。

如果没有要清理的日志，您还可以设置时间（以毫秒为单位），以将清理器放在待机中：

# ...
log.cleaner.backoff.ms=15000
# ...

例如，在下图中，特定消息键的最新消息（具有最高偏移量）被保留到压缩点。如果保留点上有任何记录，则它们会被删除。在这种情况下，压缩过程将删除所有重复。

图 3.3. 日志保留点和压缩点

当使用紧凑策略和日志清理器时，在 Kafka 中处理主题日志，请考虑优化内存分配。

您可以使用 deduplication 属性(dedupe.buffer.size)微调内存分配，它决定了为所有日志清理任务分配的内存总量。另外，您可以通过通过 buffer.load.factor 属性定义百分比来建立最大内存用量限制。

# ...
log.cleaner.dedupe.buffer.size=134217728
log.cleaner.io.buffer.load.factor=0.9
# ...

每个日志条目都使用 24 字节，以便您可以处理缓冲区可以在单一运行中处理多少个日志条目，并相应地调整设置。

如果要减少日志清理时间，请考虑增加日志清理线程数量：

# ...
log.cleaner.threads=8
# ...

如果您遇到 100% 磁盘带宽使用情况的问题，您可以节流日志清理 I/O，以便读/写操作的总和小于执行操作的磁盘功能的指定双值：

# ...
log.cleaner.io.max.bytes.per.second=1.7976931348623157E308
# ...

消息的默认批处理大小为 1MB，这是大多数用例中最大吞吐量的最佳选择。Kafka 可以在减少吞吐量下容纳更大的批处理，假设有足够的磁盘容量。

大型消息大小以四种方式处理：

建议使用基于参考的消息和消息压缩选项，并涵盖大多数情况。对于这些选项，必须小心才能避免引入性能问题。

压缩还会在生产者和解压缩开销上增加额外的处理开销，但在批处理中包含更多的数据，因此当消息数据压缩良好时，吞吐量通常很有用。

将生成者压缩与批处理大小的微调相结合，以促进最佳吞吐量。使用指标有助于量化所需的平均批处理大小。

图 3.4. 基于参考的消息传递流

当消息传递需要更多行程时，端到端延迟会增加。这个方法的另一个显著缺陷是，在清理 Kafka 消息时，外部系统中没有自动清理数据。混合方法是仅将大型消息发送到数据存储并直接处理标准化消息。

如果消息太大，则生成客户端应用必须序列化，然后对数据进行阻塞。然后，生成者使用 Kafka ByteArraySerializer，或者在发送前再次序列化每个块。消费者跟踪消息和缓冲区块，直到它有完整的消息。消耗客户端应用程序接收块，这些块在进行序列化前被编译。根据每个块消息集合的第一个或最后一个块的偏移量，将完整消息发送到消耗应用程序的剩余部分。针对偏移元数据检查成功交付完整消息，以避免重新平衡期间重复。

图 3.5. 内联消息传递流

内联消息传递在消费者端具有性能开销，因为需要缓冲，特别是在并行处理一系列大型消息时。大型消息的块可能会变为交集，因此如果缓冲区中另一个大消息的块不完整，则无法提交消息的所有区块。因此，通常通过持久保留消息块或实施提交逻辑来支持缓冲。

代理将拒绝任何大于 message.max.bytes 设置的限制的消息。producers 的缓冲区大小(max.request.size)和消费者(message.max.bytes)必须能够容纳更大的消息。

通常，建议不要设置显式清除阈值，并让操作系统使用默认设置执行后台清除。分区复制提供比写入任何单个磁盘更高的数据持久性，因为失败的代理可以从其同步的副本中恢复。

log flush 属性控制缓存的消息数据定期写入磁盘。调度程序以毫秒为单位指定日志缓存上的检查频率：

# ...
log.flush.scheduler.interval.ms=2000
# ...

您可以根据消息保留在内存中的最长时间以及日志中的最大信息数量来控制清除的频率，然后再写入磁盘：

# ...
log.flush.interval.ms=50000
log.flush.interval.messages=100000
# ...

flushes 之间的等待时间包括进行检查的时间以及执行刷新前指定的间隔。增加清除的频率可能会影响吞吐量。

如果您使用应用程序清除管理，如果您正在使用更快速的磁盘，则设置较低冲刷阈值可能适当。

分区可以在代理之间复制，以进行容错。对于给定分区，一个代理被选为 leader，并处理所有生成请求（写入日志）。在领导失败时，在其他代理中，分区会遵循其他代理为数据可靠性复制分区数据。

followers 通常不提供客户端，但 机架 配置允许消费者在 Kafka 集群跨越多个数据中心时消耗来自最接近的副本的消息。followers 仅操作从分区领导机复制消息，并允许在领导机失败时进行恢复。恢复需要同步的后续程序。遵循者通过向领导发送获取请求来保持同步，这将按顺序将消息返回到后续消息。如果已在领导消息中发现最近提交的消息，则后续者被视为同步。领导机通过查看后续程序请求的最后一个偏移来检查。不同步的后续者通常不符合领导机失败，除非允许未清理的领导选举机制。

您可以在考虑不同步前调整滞后时间：

# ...
replica.lag.time.max.ms=30000
# ...

Lag time 对时间设置了一个上限，以将消息复制到所有同步副本以及生成者必须等待确认的时间。如果后续程序无法获取请求，并捕获指定滞后时间的最新消息，则会从同步的副本中删除。您可以更快地减少检测失败的副本的时间，但这样做可能会增加无法同步的后续者数量。正确的滞后时间值取决于网络延迟和代理磁盘带宽。

当领导分区不再可用时，会选择其中一个同步副本作为新的领导。分区副本列表中的第一个代理称为 首选 领导。默认情况下，根据定期检查领导发行版，为自动分区领导重新平衡启用 Kafka。也就是说，Kafka 会检查首选领导是否为 当前的 领导。重新平衡可确保领导在代理和代理间均匀分布，不会超载。

您可以使用 Cruise Control for AMQ Streams 将副本分配找出在集群中平均平衡负载的代理。其计算需要考虑领导和后续者所经历的不同负载。失败的领导会影响 Kafka 集群的平衡，因为剩余的代理会获得领导额外分区的额外工作。

对于 Cruise Control 发现的分配，实际上，分区需要被首选领导。Kafka 可以自动确保首选领导（在可能的情况下），根据需要更改当前的领导。这样可确保集群处于 Cruise Control 找到的均衡状态。

您可以在重新平衡检查前控制重新平衡检查的频率，以及代理允许的最大 imbalance 百分比。

#...
auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
#...

代理的百分比 leader imbalance 是当前代理为当前领导的分区数量和首选领导的分区数量之间的比例。您可以将百分比设置为 0，以确保首选领导一直被选择，假设它们处于同步状态。

如果检查重新平衡需要更多控制，您可以禁用自动重新平衡。然后，您可以选择何时使用 kafka-leader-election.sh 命令行工具触发重新平衡。

对同步副本的领导选举被视为干净，因为它不会丢失数据。这是默认发生的情况。但是，如果没有同步的副本在领导方面需要什么？或许，ISR （同步副本）仅在领导磁盘结束时包含领导机。如果没有设置最少的 in-sync 副本数量，并且硬盘驱动器无法同步分区领导机时，数据将会丢失。不仅如此，新的领导产品也无法被选举，因为没有同步的跟随者。

您可以配置 Kafka 如何处理领导失败：

# ...
unclean.leader.election.enable=false
# ...

默认情况下，不干净的领导选举机制被禁用，这意味着不同步的副本无法成为领导。使用干净的领导选举机制时，如果在旧领导丢失时没有其他代理，则 Kafka 会在消息写入或读取前等待该领导机在线。不干净的领导选举机制意味着不同步的副本可能会成为领导机，但您会面临丢失消息的风险。您做出的选择取决于您的要求是否优先使用可用性或持久性。

您可以在主题级别覆盖特定主题的默认配置。如果您无法承担数据丢失的风险，请保留默认配置。

对于加入新消费者组的用户，您可以添加一个延迟，以避免对代理进行不必要的重新平衡：

# ...
group.initial.rebalance.delay.ms=3000
# ...

延迟是协调器等待成员加入的时间长度。延迟时间越长，所有成员都将一次加入并避免重新平衡的可能性。但是，延迟也会阻止组消耗到周期结束为止。

每个消费者都需要 connection 和 deserializer 属性。通常，为跟踪添加客户端 ID 是不错的做法。

在消费者配置中，无论后续配置是什么：

# ...
bootstrap.servers=localhost:9092 

key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  

value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  

client.id=my-client 

group.id=my-group-id 

# ...

用户组共享一个通常由来自给定主题的一个或多个制作者生成的大型数据流。消费者使用 group.id 属性分组，允许消息分散到成员中。组中的一个消费者被选为领导机，决定如何将分区分配给组中的消费者。每个分区只能分配给一个消费者。

如果您还没有作为分区的用户数量，您可以通过添加具有相同 group.id 的更多消费者实例来扩展数据消耗。在组中添加比分区更多的消费者将有助于吞吐量，但这意味着在待机上有消费者能够停止工作。如果您可以使用较少的使用者达到吞吐量目标，您可以节省资源。

同一消费者组中的消费者发送偏移提交和心跳到同一代理。因此组中消费者的数量越大，代理上的请求负载越高。

# ...
group.id=my-group-id 

# ...

Kafka 代理从请求代理从主题、分区和偏移位置列表中发送消息的用户接收请求。

消费者按照提交到代理的顺序在单个分区中观察信息，这意味着 Kafka 只 为单一分区中的消息提供排序保证。相反，如果消费者消耗来自多个分区的消息，则不同分区中消息的顺序与消费者观察到的消息顺序不一定反映了发送它们的顺序。

如果您希望严格排序来自一个主题的消息，请为每个消费者使用一个分区。

控制客户端应用程序调用 KafkaConsumer.poll （） 时返回的消息数量。

使用 fetch.max.wait.ms 和 fetch.min.bytes 属性来增加由 Kafka 代理使用者获取的最小数据量。基于时间的批处理使用 fetch.max.wait.ms 进行配置，并且基于大小的批处理则使用 fetch.min.bytes 来配置。

如果消费者或代理中的 CPU 使用率很高，这可能是因为消费者有太多请求。您可以调整 fetch.max.wait.ms 和 fetch.min.bytes 属性，以便在较大的批处理中交付较少的请求和信息。通过调整更高，吞吐量会降低延迟。如果生成的数据量较低，您也可以调整更高的数据。

例如，如果您将 fetch.max.wait.ms 设置为 500ms，并且 fetch.min.bytes 设为 16384 字节，当 Kafka 从消费者收到获取请求时，它将在达到第一个阈值时响应。

相反，您可以调整 fetch.max.wait.ms 和 fetch.min.bytes 属性，以提高端到端延迟。

# ...
fetch.max.wait.ms=500 

fetch.min.bytes=16384 

# ...

fetch.max.bytes 属性设置一次从代理获取的数据量的最大限制（以字节为单位）。

max.partition.fetch.bytes 会以字节为单位设置每个分区返回的最大限制，每个分区必须始终大于代理或主题配置中设置的 max.message.bytes 字节数。

客户端可消耗的最大内存量大约计算为：

NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes

如果内存用量可以容纳它，您可以增加这两个属性的值。通过允许每个请求中的更多数据，随着获取请求减少，延迟会得到提高。

# ...
fetch.max.bytes=52428800 

max.partition.fetch.bytes=1048576 

# ...

Kafka auto-commit 机制允许 使用者自动提交消息的偏移。如果启用，消费者将从轮询代理接收的偏移以 5000ms 间隔提交。

auto-commit 机制比较方便，但它引入了数据丢失和重复的风险。如果消费者获取并转换了很多消息，但系统会在执行 auto-commit 时与消费者缓冲区中处理的消息崩溃，则该数据会丢失。如果系统在处理消息后崩溃，但在执行 auto-commit 前，数据会在重新平衡后在另一个消费者实例上重复。

auto-committing 可以避免仅在下一次轮询到代理或消费者关闭前处理所有消息时数据丢失。

要最小化数据丢失或重复的可能性，您可以将 enable.auto.commit 设置为 false，并开发您的客户端应用程序来更好地控制提交偏移。或者，您可以使用 auto.commit.interval.ms 来减少提交之间的间隔。

# ...
enable.auto.commit=false 

# ...

通过将 enable.auto.commit 设置为 false，您可以在 执行所有 处理后提交偏移，并且消息已被使用。例如，您可以设置应用程序来调用 Kafka commitSync 和 commitAsync 提交 API。

commitSync API 在从轮询返回的消息批处理中提交偏移量。完成批处理中的所有消息后，会调用 API。如果使用 commitSync API，则应用程序不会轮询新消息，直到批处理中的最后一个偏移提交为止。如果这对吞吐量造成负面影响，您可以更频繁地提交，也可以使用 commitAsync API。commitAsync API 不会等待代理响应提交请求，而是在重新平衡时创建更多重复的风险。常见方法是将应用程序中的提交 API 与在关闭消费者或重新平衡前使用的 commitSync API 相结合，以确保最终提交成功。

# ...
enable.auto.commit=false
isolation.level=read_committed 

# ...

使用 session.timeout.ms 和 heartbeat.interval.ms 属性配置从消费者组中消费者故障检查和恢复的时间。

session.timeout.ms 属性指定消费者内消费者在被视为不活跃前与代理联系的最大时间，并在组中的活跃消费者间触发 重新平衡。当组重新平衡时，分区会被重新分配给组的成员。

heartbeat.interval.ms 属性指定消费者组协调器的 heartbeat 检查间隔（以毫秒为单位），以指示消费者处于活跃状态并连接。heartbeat 间隔必须较低，通常由第三个，而不是会话超时间隔。

如果您设置 session.timeout.ms 属性较低，则之前检测到失败的消费者，重新平衡可能会更快进行。但是，请小心地设置超时，以便代理无法及时接收心跳，并触发不必要的重新平衡。

减少心跳间隔可减少意外重新平衡的几率，但更频繁的心跳会增加代理资源的开销。

使用 auto.offset.reset 属性来控制消费者没有提交偏移时的行为方式，或者提交的偏移不再有效或删除。

假设您首次部署使用者应用，并且从现有的主题读取消息。由于第一次使用 group.id，因此 __consumer_offsets 主题不包含此应用的任何偏移信息。新应用可以开始处理日志开始的所有现有消息，或者仅处理新消息。默认重置值为 latest （从分区末尾开始），因此缺少一些消息。为避免数据丢失，但要增加处理量，将 auto.offset.reset 设置为 earliest 以在分区的头部开始。

另外，请考虑使用 最早 的选项来避免在为代理配置偏移保留周期(offsets.retention.minutes)时丢失消息。如果消费者组或独立消费者不活跃，并在保留期间提交没有偏移，则之前提交的偏移将从 __consumer_offsets 中删除。

# ...
heartbeat.interval.ms=3000 

session.timeout.ms=45000 

auto.offset.reset=earliest 

# ...

如果在单个获取请求中返回的数据量较大，则在消费者处理数据前可能会出现超时。在这种情况下，您可以降低 max.partition.fetch.bytes 或增加 session.timeout.ms。

在组中活跃消费者之间的分区重新平衡是：

很明显，这个过程会增加服务的停机时间，特别是在消费者组集群滚动重启后重复发生时。

在这种情况下，您可以使用 静态成员资格 的概念来减少重新平衡数量。重新平衡在消费者组成员之间平均分配主题分区。静态成员资格使用持久性，以便在会话超时后重启过程中识别消费者实例。

消费者组协调器可以使用使用 group.instance.id 属性指定的唯一 id 来识别新的消费者实例。在重启过程中，消费者被分配一个新的成员 ID，但作为静态成员，它将继续使用相同的实例 ID，并且进行相同的主题分区分配。

如果消费者应用程序没有调用至少轮询每个 max.poll.interval.ms 毫秒，则消费者被视为失败，从而导致重新平衡。如果应用程序无法处理从轮询时返回的所有记录，您可以使用 max.poll.interval.ms 属性指定从消费者轮询新消息之间的间隔（毫秒）。或者，您可以使用 max.poll.records 属性设置从消费者缓冲区返回的记录数量的最大限制，允许您的应用程序处理 max.poll.interval.ms 限值中的记录数量。

# ...
group.instance.id=UNIQUE-ID 

max.poll.interval.ms=300000 

max.poll.records=500 

# ...

每个制作者都需要 connection 和 serializer 属性。通常，最好为跟踪添加客户端 ID，并使用生成者压缩来减小请求的批处理大小。

在基本制作者配置中：

# ...
bootstrap.servers=localhost:9092 

key.serializer=org.apache.kafka.common.serialization.StringSerializer 

value.serializer=org.apache.kafka.common.serialization.StringSerializer 

client.id=my-client 

compression.type=gzip 

# ...

消息发送确认可最大程度降低消息丢失的可能性。默认情况下，使用 acks=all 设置的 acks 属性启用确认。

# ...
acks=all 

# ...

acks=all 设置提供了最强的交付保证，但它会增加生产者发送消息和接收确认之间的延迟。如果您不要求这种强保证，则 acks=0 或 acks=1 的设置不提供交付保证，或者仅确认领导副本已将记录写入其日志中。

使用 acks=all 时，领导机会等待所有同步副本确认消息发送。主题的 min.insync.replicas 配置设定了同步副本确认所需的最小所需数量。确认数量包括领导和跟着者的数量。

典型的起点是使用以下配置：

在创建主题时，您可以覆盖默认的复制因素。您还可以在主题配置的主题级别覆盖 min.insync.replicas。

AMQ Streams 在示例配置文件中使用此配置进行 Kafka 的多节点部署。

下表描述了此配置的工作方式，具体取决于复制领导副本的后续者可用性。

3 的主题复制因素会创建一个领导副本和两个后续者。在此配置中，如果单个后续者不可用，则生成者可以继续。有些延迟可能会发生 whilst 从同步副本中删除失败的代理或创建新领导。如果第二个后续人也不可用，消息发送将无法成功。领导者不会发送成功消息，而不是向制作者发送错误(而非充足的副本)。生产者引发等同的异常。通过 重试 配置，生成者可以重新发送失败的消息请求。

幂等的生成者会避免在消息发送一次时重复。ID 和序列号分配给消息，以确保发送顺序，即使出现失败情况。如果您使用 acks=all 用于数据一致性，使用 idempotency 对排序的交付有意义。默认情况下，为制作者启用 idempotency。启用 idempotency 后，您可以将并发 in-flight 请求的数量设置为最多 5 个，以便保留消息排序。

# ...
enable.idempotence=true 

max.in.flight.requests.per.connection=5 

acks=all 

retries=2147483647 

# ...

如果您选择不使用 acks=all 并因为性能成本而禁用 idempotency，请将 in-flight （未确认）请求数量设置为 1 来保留顺序。否则，在 Message-A 已写入代理后，Message- A 才会成功。

# ...
enable.idempotence=false 

max.in.flight.requests.per.connection=1 

retries=2147483647
# ...

在对单个分区进行一次写入一次，idempotence 非常有用。事务与 idempotence 一起使用时，允许跨多个分区的一次写入一次。

使用相同事务 ID 发送的事务消息只生成一次，因此只会所有都成功写入到相应的日志，或所有都没有写入。

# ...
enable.idempotence=true
max.in.flight.requests.per.connection=5
acks=all
retries=2147483647
transactional.id=UNIQUE-ID 

transaction.timeout.ms=900000 

# ...

选择 transactional.id 非常重要，以便保持事务保证。每个事务 ID 都应该用于一组唯一的主题分区。例如，可以使用主题分区名称的外部映射到事务 ID，或使用避免冲突的功能计算主题分区名称中的事务 ID 来实现。

通常，系统的要求是满足给定延迟内消息的特定吞吐量目标。例如，以每秒 500,000 条消息为目标，在 2 秒内确认消息的 95%。

您的制作者的消息语义（消息排序和持久性）可能由您的应用程序的要求定义。例如，您可能没有选择使用 acks=0 或 acks=1，而不破坏一些重要属性或应用程序提供的保证。

代理重启会对高百分比统计产生重大影响。例如，在较长的时间内，代理重启过程中会造成 99 百分比的延迟。在设计基准测试中或比较基准测试与生产环境中看到的性能号进行比较时，这值得考虑。

根据您的目标，Kafka 提供了很多配置参数和技术，用于调优生成者性能以获得吞吐量和延迟。

点 send （） 方法返回。因此，阻止了时间 send （） 是由以下决定的：

批处理将保留在队列中，直到出现以下情况之一：

查看批处理和缓冲区的配置，以减少 send （） 阻止对延迟的影响。

# ...
linger.ms=100 

batch.size=16384 

buffer.memory=33554432 

# ...

您还可以通过编写自定义分区来替换默认值，将消息定向到指定的分区。

# ...
delivery.timeout.ms=120000 

partitioner.class=my-custom-partitioner 

# ...