Kafka 配置调整

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

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

num.partitions:1

num.partitions:1

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

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 个主题，并将所需最小同步副本设置为 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 复制因素配置，其中至少 2 个同步副本用于这个主题，这意味着 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 1
queued.max.requests=500 2
num.io.threads=8 3
num.recovery.threads.per.data.dir=4 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 在主题级别上覆盖这些设置。无论您需要降低还是提升这些值，都取决于删除网段的策略。较大的大小表示活动片段包含更多信息，而且经常进行推广。细分市场也成为对删除条件的资格。

您可以设置基于时间或基于大小的日志保留和清理策略，以便日志保持可管理。根据您的要求，您可以使用日志保留配置来删除旧的片段。如果使用日志保留策略，则在达到保留限制时会删除非活跃日志片段。删除旧的片段会绑定日志所需的存储空间，因此不会超过磁盘容量。

对于基于时间的日志保留，您可以根据小时、分钟和毫秒设置保留周期。保留周期基于在网段中附加信息的时间。

毫秒毫秒的配置优先于分钟，而优先级超过小时。默认情况下，minutes 和 毫秒的配置是 null，但这三个选项提供了对您要保留的数据的大量控制级别。首选应以毫秒为单位指定，因为它是唯一可动态更新的三个属性之一。

...
...

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

如果 log.retention.ms 设为 -1，则不会应用时间限制来记录保留，因此所有日志都会被保留。通常不建议监控磁盘用量，但通常不建议使用 -1 设置，因为它可能会导致完整磁盘出现问题，这可能会很难重新处理。

对于基于大小的日志保留，您需要设置一个最大日志大小（日志中所有片段）以字节为单位：

...
...

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

换句话说，日志通常将有大约 log.retention.bytes/log.segment.bytes 段，当它达到稳定状态。当达到最大日志大小时，会删除旧的片段。

使用最大日志大小的一个潜在问题是，它不会考虑将消息附加到网段的时间。您可以使用基于时间和基于时间的日志保留来使您的清理策略达到所需的平衡。达到哪个阈值首先触发清理。

如果要在从系统中删除片段文件前添加时间延迟，您可以使用 log.segment.delete.delay.ms 在代理级别或 file.delete.delay.ms 中针对特定主题的 log.segment.delete.delay.ms 添加延迟。

...
...

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

删除旧日志数据的方法由 日志清理 配置决定。

默认情况下，代理启用了 log cleaner：

...
...

# ...
log.cleaner.enable=true
# ...

如果您使用日志压缩清理策略，则需要启用日志清理器。您可以在 topic 或 broker 级别设置清理策略。broker-level 配置是没有策略集的主题的默认设置。

您可以设置策略来删除日志、紧凑日志或两者都：

...
...

# ...
log.cleanup.policy=compact,delete
# ...

删除策略 与使用数据保留策略来管理日志对应。当数据不需要永久保留数据时，它很合适。紧凑 策略保证保留每条消息键的最新消息。日志压缩非常适合消息值可更改，而您想要保留最新更新。

如果清理策略被设置为删除日志，则会根据日志保留限制删除旧的片段。否则，如果没有启用日志清理器，且没有日志保留限制，日志将继续增大。

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

图 3.1. 在压缩前显示键值写入使用偏移位置的写入

使用密钥来识别信息，Kafka 压缩为特定消息键保留最新的信息（使用最高偏移量），最终丢弃了具有相同密钥的先前消息。换句话说，其最新状态的消息始终可用，当日志清理器运行时，该特定消息的任何过期记录都将最终删除。您可以将消息恢复回以前的状态。

记录会保留其原始偏移，即使周围的记录被删除。因此，tail 可能会具有非协调误差。当消耗在 tail 中不再可用的偏移值时，会找到带有下一较高偏移的记录。

图 3.2. 压缩后的日志

如果只选择紧凑策略，则日志仍然可以成为任意大。在这种情况下，您可以将策略设置为紧凑 并删除 日志。如果您选择紧凑并删除日志，则首先删除日志数据，在日志头部使用键删除记录。之后，删除日志保留阈值前的数据。

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

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

...
...

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

调整与日志保留设置相关的日志保留检查间隔。较小的保留大小可能需要更频繁地检查。

清理的频率应该足以管理磁盘空间，但通常不会影响主题的性能。

如果没有要清理的日志，您也可以以毫秒为单位设置一个时间，以将清理设置为待机：

...
...

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

如果选择删除旧的日志数据，您可以在清除前以毫秒为单位设置保留删除的数据的周期：

...
...

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

删除的数据保留周期可让时间注意到数据已消失，然后再被删除。

要删除与特定密钥相关的所有消息，生产者可以发送 tombstone 消息。tombstone 具有 null 值，并充当一个标记来告知消费者值被删除。在压缩后，只有 tombstone 会被保留，这必须在足够长的时间段内才可知道该消息已被删除。当旧的消息被删除时，没有值，则会从分区中删除 tombstone 密钥。

还有许多其他配置设置与日志清理相关，但特别重要是内存分配。

deduplication 属性指定在所有日志清理线程之间清理的总内存。您可以对通过缓冲区负载因素使用的内存百分比设置上限。

...
...

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

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

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

...
...

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

如果您在 100% 磁盘带宽使用量时遇到问题，您可以减慢日志清理 I/O，以便 read/write 操作的总和小于执行操作的磁盘的双重值：

...
...

# ...
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 之间的等待包括执行清除前检查的时间以及清除前的指定间隔。增加刷新的频率可能会影响吞吐量。

如果使用应用程序冲刷管理，如果您使用的是更快速的磁盘，则设置小于冲刷阈值。

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

遵循者通常不为客户端提供服务，但 机架 配置允许消费者在 Kafka 集群跨越多个数据中心时使用来自最接近的副本的消息。仅限后续者只从分区领导设备复制信息，并允许恢复失败。恢复需要一个 in-sync follower。跟进者通过将 get 请求发送到领导请求来保持同步，按顺序将消息返回到后续程序。如果遇到最近提交的消息在领导上的最新提交的消息，则跟进者就被视为同步。领导者通过查看后续者请求的最后一个偏移来对此进行检查。除非允许未干净的领导选举机制，否则，直到同步后续程序通常不符合领导机 后才应具备资格。

您可以在后续者视为同步范围前调整 lag 时间：

...
...

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

lag time 定时限于将消息复制到所有 in-sync 副本，以及生产者必须等待被确认的时间。如果后续程序未能生成 fetch 请求，并在指定的 lag 时间内获取最新的消息，则会从同步副本中删除。您可以缩短早检测失败的副本的滞后时间，但这样做可能会增加无法同步的后续者数量。正确的 lag 时间值取决于网络延迟和代理磁盘带宽。

当领导分区不再可用时，选择其中一个 in-sync 副本作为新领导。分区副本列表中的第一个代理称为 首选 领导者。默认情况下，基于定期检查领导分布，默认启用 Kafka 以进行自动分区领导重新平衡。也就是说，Kafka 会检查首选领导是 当前的 领导人。重新平衡可确保领导人在代理和代理之间均匀分布。

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

对于由 Cruise Control 找到的分配，需要实际平衡分区是首选领导分区所致的。Kafka 可以自动确保使用首选领导者（可能），根据需要更改当前的领导人。这样可确保集群保持在 Cruise Control 找到的 balanced 状态。

在触发重新平衡前，您可以控制重新平衡检查的频率（以秒为单位），以及对代理允许的 imbalance 百分比。

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

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

代理的领导值是当前代理是当前领导的分区数和首选领导分区数量之间的比例。您可以将百分比设定为零，以确保始终选择首选领导（假设它们处于同步状态）。

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

对同步副本的领导选举被视为干净，因为它保证不会丢失数据。这是默认的情况。但是，如果没有内部同步副本来开展领导？或许，ISR （同步副本）仅在领导磁盘中断时才会包含领导机。如果没有设置最少的 in-sync 副本，当硬盘驱动器出现故障时，没有遵循者与分区领导设备同步，则数据已经丢失。不仅如此，新的领导产品也无法被选举，因为没有同步的跟随者。

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

...
...

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

默认情况下不干净的领导选举机制被禁用，这意味着内存不足的副本无法成为领导者。使用干净的领导选举机制时，如果没有其他代理在 ISR 中丢失，则 Kafka 会等待领导人恢复在线，然后才能写入或读取消息。取消清除领导选举意味着不同步的副本可能会成为领导机，但您将面临丢失消息的风险。您做出的选项取决于您的要求是否偏离可用性。

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

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

...
...

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

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

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

在消费者配置中，任何后续配置都相关：

...
...

# ...
bootstrap.servers=localhost:9092 1
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  2
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  3
client.id=my-client 4
group.id=my-group-id 5
# ...

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

如果还没有拥有数量的消费者作为分区，您可以通过添加更多具有相同 组.id 的消费者实例来扩展数据消耗。将更多消费者添加到组中，而不是分区的吞吐量，但它意味着有消费者处于待机状态，应该可以停止运行。如果可以通过较少的消费者实现吞吐量目标，您可以节约资源。

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

...
...

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

Kafka 代理接收来自消费者的请求，请求代理从主题、分区和偏移位置列表中发送信息。

消费者按照它们提交到代理的顺序观察单个分区中的消息，这意味着 Kafka 仅为单个分区 中的消息提供排序保障。相反，如果消费者消耗来自多个分区的消息，如消费者观察到的不同分区中的消息顺序并没有反映它们所发送的顺序。

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

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

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

如果使用者或代理中的 CPU 使用率很高，则可能是因为来自消费者的请求太多。您可以调整 fetch.max.wait.ms 和 fetch.min.bytes 属性，以便请求和消息在较大的批处理中交付较少。通过调整较高情况，可以提高吞吐量，同时将一些成本降低到延迟。如果生成的数据量较低，您也可以调整更高。

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

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

...
...

# ...
fetch.max.wait.ms=500 1
fetch.min.bytes=16384 2
# ...

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

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

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

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

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

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

...
...

# ...
fetch.max.bytes=52428800 1
max.partition.fetch.bytes=1048576 2
# ...

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

自动提交机制很方便，但会带来数据丢失和重复的风险。如果消费者已获取并转换了很多消息，但如果执行自动提交时，系统会因为消费者缓冲区中处理的消息崩溃，则数据将会丢失。如果在处理消息后系统崩溃，但在执行 auto-commit 前，数据会在重新平衡后在另一使用者实例上重复。

只有在下一次轮询代理或消费者关闭前处理所有消息时，自动提交可以避免数据丢失。

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

...
...

# ...
enable.auto.commit=false 1
# ...

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

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

...
...

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

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

session.timeout.ms 属性指定消费者组中的使用者可以在被视为不活跃前与代理联系，并在组中在活动消费者之间触发 重新平衡 的时间，以毫秒为单位。当组重新平衡时，分区会被重新分配给组的成员。

heartbeat.interval.ms 属性指定 heartbeat 检查到消费者组协调器之间的时间间隔（毫秒），以表示使用者处于活动状态和连接。heartbeat 间隔必须降低，通常要小于会话超时间隔。

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

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

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

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

另外，请考虑使用 最早的 选项来避免在偏移保留周期(偏移.retention.minutes)被终止时丢失消息。如果消费者组或独立使用者在保留期间没有激活，并且提交之前提交的偏移量从 __consumer_offsets 中删除。

...
...

# ...
heartbeat.interval.ms=3000 1
session.timeout.ms=45000 2
auto.offset.reset=earliest 3
# ...

如果单个提取请求中返回的数据量较大，则可能需要在消费者处理它前发生超时。在这种情况下，您可以减小 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 1
max.poll.interval.ms=300000 2
max.poll.records=500 3
# ...

每个制作者都需要连接和序列化属性。通常，最好添加一个客户端 ID 进行跟踪，并使用制作者上的压缩来减少请求的批处理大小。

在基本的制作者配置中：

...
...

# ...
bootstrap.servers=localhost:9092 1
key.serializer=org.apache.kafka.common.serialization.StringSerializer 2
value.serializer=org.apache.kafka.common.serialization.StringSerializer 3
client.id=my-client 4
compression.type=gzip 5
# ...

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

...
...

# ...
acks=all 1
# ...

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

使用 acks=all 时，领导机会等待所有同步副本确认消息发送。主题的 min.insync.replicas 配置设置所需的最小数量 in-sync 副本确认。承认的编号包括领导人和后续人。

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

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

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

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

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

幂等的生产者可避免在消息交替发送时重复。ID 和序列号被分配到消息，以确保交付顺序，即使在出现故障时也是如此。如果您将 acks=all 用于数据一致性，使用幂等性对订购的发送有意义。默认情况下，为制作者启用幂等性。启用 idempotency 时，您可以将并发 in-flight 请求数量设置为最多 5 个要保留的消息排序。

...
...

# ...
enable.idempotence=true 1
max.in.flight.requests.per.connection=5 2
acks=all 3
retries=2147483647 4
# ...

如果您选择使用 acks=all 和 disable idempency，因为性能成本，请将非航班（未确认）的请求数量设置为 1 以保留排序。否则，当 Message-B 已被写入代理后，Message-A 才会成功。

...
...

# ...
enable.idempotence=false 1
max.in.flight.requests.per.connection=1 2
retries=2147483647
# ...

在向单个分区写入一个分区后，Idempotence 非常有用。事务（与幂等使用）一起使用时，允许在多个分区间写入一次。

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

...
...

# ...
enable.idempotence=true
max.in.flight.requests.per.connection=5
acks=all
retries=2147483647
transactional.id=UNIQUE-ID 1
transaction.timeout.ms=900000 2
# ...

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

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

您的制作者的消息传递语义（消息排序和持续时间）可能由您的应用程序的要求定义。例如，您不能选择使用 acks=0 或 acks=1 选项，而无需破坏某些重要属性或保证应用程序提供。

代理重启会对高百分比统计造成显著影响。例如，在较长时间内，代理重启的行为是 99%ile 延迟的划分。在设计基准或比较了生产中的性能编号时，可以考虑这一点。

根据您的目标，Kafka 提供了多个配置参数和技巧，用于针对吞吐量和延迟调整制作者性能。

在哪个指向 send （） 方法返回。因此，时间 send （） 由以下决定：

批处理将保留在队列中，直到发生以下之一：

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

...
...

# ...
linger.ms=100 1
batch.size=16384 2
buffer.memory=33554432 3
# ...

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

...
...

# ...
delivery.timeout.ms=120000 1
partitioner.class=my-custom-partitioner 2

# ...