kafka官方文档中文翻译（kafka参数解释）

入门

1.1简介

kafka™是一个分布式流媒体平台。这到底意味着什么？

我们认为流媒体平台具有三个关键功能：

它允许您发布和订阅记录流。在这方面，它类似于消息队列或企业消息系统。
它允许您以容错方式存储记录流。
它允许您在记录发生的时期处理记录。

Kafka有什么优点？

它用于两大类应用程序：

构建可靠地在系统或应用程序之间获取数据的实时流数据管道
构建变换或响应数据流的实时流应用程序

要了解Kafka如何做这些事情，让我们从下而上地研究和探索Kafka的功能。

首先几个概念：

Kafka作为一个集群在一个或多个服务器上运行。
kafka集群按Topic存储的分类数据流。
每个记录由一个键，一个值和一个时间戳组成。

Kafka有四个核心API：

生产者API允许应用程序发布流记录到一个或多个kafka的主题。
消费者API允许应用程序订阅一个或多个主题和处理所产生的对他们的记录流。
流API允许应用程序作为流处理器，从一个消费输入流一个或多个主题，并产生一个输出流到一个或多个输出主题，高效地传输这个输入流到输出流。
连接器API允许构建和运行可重复使用的生产者或消费者连接kafka主题到现有的应用程序或数据系统。例如，关系数据库的连接器可能捕获对表的每个更改。

在kafka客户端和服务器之间的通信以简单的，高性能的，语言无关完成TCP协议。此协议版本化，并保持与旧版本的向后兼容性。我们对kafka提供了一个Java客户端，但是客户端在多种语言中都可以使用。

主题和日志

让我们首先深入Kafka提供的记录流的核心抽象 - 主题。

主题是发布记录的类别或Feed名称。主题在Kafka总是多用户; 也就是说，主题可以具有零个，一个或多个订阅订阅其的数据的消费者。

对于每个主题，Kafka集群维护一个分区日志，如下所示：

每个分区是一个有序的，不可变的记录序列，不断地附加到结构化提交日志。在分区中的记录是所谓每个指派顺序ID号的偏移量唯一地标识该分区中的每个记录。

Kafka集群保留所有已发布的记录，无论它们是否已使用可配置的保留期。例如，如果保留策略设置为两天，则在发布记录后的两天内，可以使用该记录，之后将被丢弃以释放空间。Kafka的性能在数据大小方面是有效的，因此长时间存储数据不是问题。

事实上，每个消费者保留的唯一元数据是消费者在日志中的偏移或位置。这种偏移由消费者控制：通常消费者在读取记录时线性地提前其偏移，但是事实上，由于位置由消费者控制，所以它可以按照喜欢的任何顺序来消费记录。例如，消费者可以重置到较旧的偏移以重新处理来自过去的数据或者跳到最近的记录并开始从“现在”消费。

这些功能的结合意味着Kafka消费者非常便宜 - 他们可以来来去去，对群集或其他消费者没有太大的影响。例如，您可以使用我们的命令行工具“拖动”任何主题的内容，而无需更改任何现有用户使用的内容。

日志中的分区有几个目的。首先，它们允许日志扩展到适合单个服务器的大小。每个单独的分区必须适合托管它的服务器，但一个主题可能有许多分区，因此它可以处理任意数量的数据。第二，它们作为并行性的单位 - 更多的是在一点。

分配

日志的分区分布在Kafka集群中的服务器上，每个服务器处理数据并请求共享分区。每个分区都跨越可配置数量的服务器进行复制，以实现容错。

每个分区具有用作“领导者”的一个服务器和充当“跟随者”的零个或多个服务器。领导者处理分区的所有读取和写入请求，而关注者被动地复制领导者。如果领导失败，其中一个追随者将自动成为新的领导者。每个服务器作为其一些分区的领导者和为其他分区的追随者，所以负载在集群内是平衡的。

生产者

生产者将数据发布到他们选择的主题。生产者负责选择哪个记录分配给主题内的哪个分区。这可以以循环方式完成以简单地平衡负载，或者它可以根据一些语义分区函数（例如基于记录中的一些密钥）来完成。更多关于使用分区的一秒钟！

消费者

消费者标榜自己与消费群的名称，并发布到一个话题每个记录每个订阅用户组内传送到一个消费者的实例。消费者实例可以在单独的进程中或在单独的机器上。

如果所有消费者实例具有相同的消费者组，则记录将有效地在消费者实例上进行负载平衡。

如果所有消费者实例具有不同的消费者组，则每个记录将被广播到所有消费者进程。

两个服务器Kafka集群托管四个分区（P0-P3）与两个消费者组。消费者组A有两个消费者实例，组B有四个。

然而，更常见的是，我们发现主题具有少量的消费者组，每个“逻辑用户”一个。每个组由用于可伸缩性和容错的许多消费者实例组成。这只是发布 - 订阅语义，其中订户是消费者的集群而不是单个进程。

在Kafka中实现的方式是通过划分消费者实例上的日志中的分区，使得每个实例在任何时间点是分区的“公平共享”的独占消费者。维护组中成员资格的过程由Kafka协议动态处理。如果新实例加入组，则它们将从组的其他成员接管一些分区; 如果实例死机，其分区将分发到其余实例。

卡夫卡只提供了记录的总订单中的一个分区，而不是一个主题的不同分区之间。每分区排序结合按键分区数据的能力对于大多数应用程序是足够的。但是，如果您需要对记录进行总排序，则可以使用只有一个分区的主题来实现，但这将意味着每个用户组只有一个使用者进程。

保证

在高级Kafka提供以下保证：

生产者发送到特定主题分区的消息将按照它们发送的顺序附加。也就是说，如果记录M1由与记录M2相同的生成器发送，并且M1首先发送，则M1将具有比M2更低的偏移并且在日志中较早出现。
消费者实例按记录存储在日志中的顺序查看记录。
对于具有复制因子N的主题，我们将允许最多N-1个服务器故障，而不会丢失提交到日志的任何记录。

有关这些保证的更多详细信息，请参见文档的设计部分。

Kafka作为一个消息系统

Kafka的流概念与传统的企业消息系统相比如何？

消息历来有两种型号：队列和发布-订阅。在队列中，消费者池可以从服务器读取并且每个记录去往其中一个; 在发布 - 订阅中，记录被广播给所有消费者。这两种模式都有其优点和缺点。排队的优势在于它允许您划分多个消费者实例上的数据处理，这样可以扩展处理。不幸的是，队列不是多用户 - 一旦一个进程读取它消失的数据。发布订阅允许您将广播数据传送到多个进程，但是没有办法缩放处理，因为每个消息都发送给每个订阅者。

Kafka中的消费者群体概念概括了这两个概念。与队列一样，使用者组允许您对一组进程（消费者组的成员）分配处理。与发布 - 订阅一样，Kafka允许您向多个用户组广播消息。

Kafka的模型的优点是每个主题都有这些属性 - 它可以扩展处理，也是多订户 - 没有必要选择一个或另一个。

Kafka也有比传统的消息传递系统更强的顺序保证。

传统队列在服务器上按顺序保留记录，并且如果多个消费者从队列中消耗，则服务器按照它们被存储的顺序发出记录。然而，尽管服务器按顺序发出记录，但是记录被异步地递送给消费者，因此他们可能在不同的消费者处乱序到达。这有效地意味着在存在并行消耗的情况下记录的排序丢失。消息传递系统通常通过具有仅允许一个进程从队列消费的“独占消费者”的概念来解决这个问题，但是当然这意味着在处理中没有并行性。

卡夫卡做得更好。通过在主题内部具有并行性的概念 - 分区 - ，Kafka能够在消费者进程池上提供排序保证和负载均衡。这通过将主题中的分区分配给消费者组中的消费者来实现，使得每个分区仅由组中的一个消费者消费。通过这样做，我们确保消费者是该分区的唯一读取器，并按顺序消耗数据。由于有许多分区，这仍然平衡许多消费者实例上的负载。但请注意，消费者组中不能有比分区更多的消费者实例。

Kafka作为存储系统

任何允许发布消息从消费消息中分离的消息队列实际上充当了正在传输消息的存储系统。Kafka的不同之处在于它是一个非常好的存储系统。

写入到Kafka的数据写入磁盘并复制以用于容错。Kafka允许生产商等待确认，以便写入被认为是完整的，直到它被完全复制，并保证即使服务器写入失败仍然持续。

磁盘结构Kafka使用Scale-well，Kafka将执行相同的，无论您在服务器上有50 KB或50 TB的持久数据。

由于认真对待存储并允许客户端控制其读取位置，您可以将Kafka视为一种专用于高性能，低延迟提交日志存储，复制和传播的特殊用途分布式文件系统。

用于流处理的Kafka

仅仅读取，写入和存储数据流是不够的，其目的是实现流的实时处理。

在Kafka中，流处理器是从输入主题获取连续数据流，对这个输入执行一些处理，并产生连续数据流到输出主题的任何东西。

例如，零售应用可以接收销售和货物的输入流，并输出从该数据计算出的重新排序和价格调整流。

可以直接使用生产者和消费者API进行简单的处理。然而，对于更复杂的转换卡夫卡提供了一个完全集成的流API。这允许构建执行不重要的处理的应用程序，其计算流的聚合或将流连接在一起。

这个工具帮助解决这种类型的应用面临的硬问题：处理乱序数据，将代码重新处理输入作为代码更改，执行状态计算等。

streams API基于Kafka提供的核心原语：它使用生产者和消费者API用于输入，使用Kafka进行有状态存储，并且在流处理器实例之间使用相同的组机制进行容错。

把碎片放在一起

这种消息传递，存储和流处理的组合可能看起来不寻常，但对于Kafka作为流式传输平台的作用至关重要。

像HDFS这样的分布式文件系统允许存储静态文件以进行批处理。有效像这样的系统允许存储和处理的历史，从过去的数据。

传统的企业邮件系统允许处理在您订阅之后到达的未来邮件。以这种方式构建的应用程序在未来数据到达时处理它。

Kafka结合了这两个功能，这种组合对于Kafka作为流应用程序和流数据流水线的平台至关重要。

通过组合存储和低延迟订阅，流应用程序可以以相同的方式处理过去和未来的数据。这是一个单一的应用程序可以处理历史，存储的数据，而不是结束时，当它到达最后一个记录，它可以保持处理作为未来的数据到达。这是包含批处理以及消息驱动应用程序的流处理的概括概念。

同样，对于流数据流水线，订阅实时事件的组合使得可以将Kafka用于非常低延迟的流水线; 但是可靠地存储数据的能力使得可以将其用于必须保证数据传送的关键数据，或者用于与仅周期性地加载数据的离线系统集成，或者可以长时间地进行维护。流处理设施使得可以在数据到达时变换数据。

有关担保，API和功能的更多信息，卡夫卡提供看剩下的文档。

1.2使用案例

这里是对几个受欢迎的用例的Apache Kafka™的描述。对于一些行动，这些领域的概述，看到这个博客帖子。

消息

Kafka很好地代替了一个更传统的消息代理。消息代理由于各种原因（用于将处理与数据生成器分离，缓冲未处理的消息等）而使用。与大多数消息传递系统相比，Kafka具有更好的吞吐量，内置分区，复制和容错功能，使其成为大规模消息处理应用程序的良好解决方案。

在我们的经验中，消息传递使用通常比较低的吞吐量，但可能需要低端到端延迟，并且通常取决于Kafka提供的强大的持久性保证。

在这一领域的卡夫卡媲美传统的邮件系统，如的ActiveMQ或RabbitMQ的。

网站活动跟踪

Kafka的原始用例是能够将用户活动跟踪管道重建为一组实时发布订阅源。这意味着网站活动（网页浏览量，搜索或用户可能采取的其他操作）被发布到中心主题，每个活动类型有一个主题。这些订阅源可用于订阅一系列用例，包括实时处理，实时监控，加载到Hadoop或离线数据仓库系统以进行离线处理和报告。

活动跟踪通常是非常高的量，因为为每个用户页面视图生成许多活动消息。

指标

Kafka通常用于操作监控数据。这涉及聚合来自分布式应用程序的统计信息，以产生操作数据的集中馈送

日志聚合

许多人使用Kafka替换日志聚合解决方案。日志聚合通常从服务器收集物理日志文件，并将它们放在中央位置（文件服务器或HDFS可能）进行处理。Kafka提取文件的详细信息，并将日志或事件数据更清晰地抽象为消息流。这允许更低延迟的处理和更容易支持多个数据源和分布式数据消耗。与诸如Scribe或Flume的以日志为中心的系统相比，Kafka提供了同样出色的性能，由于复制而提供的更强的持久性保证以及更低的端到端延迟。

流处理

Kafka的许多用户在由多个阶段组成的处理管道中处理数据，其中原始输入数据从Kafka主题消费，然后聚合，丰富或以其他方式转换为新主题以供进一步消费或后续处理。例如，用于推荐新闻文章的处理管道可以从RSS订阅源爬行文章内容并将其发布到“文章”主题; 进一步处理可以规范化或去重复该内容并且将经过清洗的文章内容发布到新主题; 最终处理阶段可能会尝试向用户推荐此内容。这样的处理流水线基于各个主题创建实时数据流的图形。在0.10.0.0开始，重量轻，但功能强大的流处理库调用卡夫卡流可在Apache的卡夫卡如上所述进行这样的数据处理。除了卡夫卡流，替代开源流处理工具包括阿帕奇风暴和阿帕奇Samza。

事件源

事件采购是应用程序的设计风格，其中状态变化会被记录为一个记录时间的有序序列。Kafka对非常大的存储日志数据的支持使其成为以这种风格构建的应用程序的极好的后端。

提交日志

Kafka可以作为一种分布式系统的外部提交日志。日志有助于在节点之间复制数据，并作为故障节点恢复其数据的重新同步机制。该日志压实卡夫卡功能有助于支持这种用法。在这种用法卡夫卡类似Apache的会计项目。

1.3快速入门

本教程假设您是新开始的，并且没有现有的Kafka™或ZooKeeper数据。由于卡夫卡控制台脚本是基于Unix和Windows平台上的不同，在Windows平台上使用

bin\windows\

，而不是

bin/

和修改脚本扩展

.bat

。

步骤1：下载代码

下载的0.10.1.0释放和解压它。

> tar-xzf kafka_2.11-0.10.1.0.tgz 
> cd kafka_2.11-0.10.1.0

步骤2：启动服务器

Kafka使用ZooKeeper，所以你需要首先启动一个ZooKeeper服务器，如果你还没有。您可以使用与kafka一起提供的便利脚本来获取快速清理的单节点ZooKeeper实例。

> bin/zookeeper-server-start.sh config/zookeeper.properties
[2013-04-22 15：01：37,495] Reading configuration from: config / zookeeper.properties（org.apache.zookeeper.server.quorum.QuorumPeerConfig）
... ...

现在启动Kafka服务器：

>bin/kafka-server-start.sh config/server.properties
[2013-04-22 15：01：47,028] INFO Verifying properties（kafka.utils.VerifiableProperties）
[2013-04-22 15：01：47,051] INFO Property socket.send.buffer.bytes overidden to 1048576（kafka.utils.VerifiableProperties）
... ...

步骤3：创建主题

让我们使用单个分区和一个副本创建一个名为“test”的主题：

> bin/kafka-topics.sh --create --zookeeperlocalhost:2181 --replication-factor 1 --partitions 1 --topic test

如果我们运行list topic命令，我们现在可以看到这个主题：

> bin/ kafka-topics.sh --list --zookeeper localhost:2181
test

或者，您也可以将经销商配置为在发布不存在的主题时自动创建主题，而不是手动创建主题。

步骤4：发送一些消息

Kafka提供了一个命令行客户端，它将从文件或标准输入接收输入，并将其作为消息发送到Kafka集群。默认情况下，每行都将作为单独的消息发送。

运行生产者，然后在控制台中键入一些消息发送到服务器。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
这是一个消息
这是另一条消息

步骤5：启动消费者

Kafka还有一个命令行消费者，将消息转储到标准输出。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
这是一条消息
这是另一个消息

如果你有上面的每个命令运行在不同的终端，那么你现在应该能够在生产者终端中键入消息，看到它们出现在消费者终端。

所有命令行工具都有其他选项; 运行没有参数的命令将更详细地显示记录它们的使用信息。

步骤6：设置多代理集群

到目前为止，我们一直在对一个单一的broker，但这没有乐趣。对于Kafka，单个代理只是一个大小为1的集群，因此除了启动几个代理实例之外没有什么变化。但只是为了感觉到它，让我们将我们的集群扩展到三个节点（仍然在我们的本地机器上）。

首先我们做一个配置文件为每个broker的（在Windows上使用

copy

命令来代替）：

> cp config/server.properties config/server-1.properties 
> cp config/server.properties config/server-2.properties

现在编辑这些新文件并设置以下属性：

config/server-1.properties：
    broker.id = 1
    listeners = PLAINTEXT://:9093
    log.dir = /tmp/kafka-logs-1

config/server-2.properties:
    broker.id = 2
    listeners = PLAINTEXT://:9094
    log.dir = /tmp/kafka-logs-2

的

broker.id

属性是集群中的每个节点的唯一且永久的名称。我们必须覆盖端口和日志目录，因为我们正在同一台机器上运行这些，我们希望保持所有代理尝试在同一端口注册或覆盖彼此的数据。

我们已经有了Zookeeper和我们的单节点启动，所以我们只需要启动两个新的节点：

> bin/kafka-server-start.sh config/server-1.properties ＆
... ...
> bin/kafka-server-start.sh config/server-2.properties ＆
... ...

现在创建一个新的主题，复制因子为3：

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

好了，但现在我们有一个集群，我们怎么知道哪个代理正在做什么？要看到运行“describe topics”命令：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
topic：my-replicated-topic partitionCount：1 ReplicationFactor：3 config：
    topic：my-replicated-topic partition：0 leader：1 Replicas：1,2,0 Isr：1,2,0

这里是输出的解释。第一行给出所有分区的摘要，每个附加行给出关于一个分区的信息。因为我们对于这个主题只有一个分区，所以只有一行。

“leader”是负责给定分区的所有读取和写入的节点。每个节点将是分区的随机选择部分的领导者。
“replicas”是复制此分区的日志的节点的列表，无论它们是否为引导者，或者即使它们当前处于活动状态。
“isr”是“同步中”副本的集合。这是副本列表的子集，其当前活动并赶上领导者。

注意，在我的示例中，节点1是主题的唯一分区的leader。

我们可以对我们创建的原始主题运行相同的命令，以查看它在哪里：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
主题：test PartitionCount：1 ReplicationFactor：1配置：
    主题：测试分区：0领导：0副本：0 Isr：0

所以没有什么惊喜 - 原来的主题没有副本，并在服务器0，我们的集群中的唯一的服务器，当我们创建它。

让我们向我们的新主题发布几条消息：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
... ...
我的测试消息1 
我的测试消息2 
^ C

现在让我们使用这些消息：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
... ...
我的测试消息1
我的测试消息2
^ C

现在让我们测试容错。broker1担任领导者，所以让我们kill了它：

> ps aux | grep server-1.properties 
7564 ttys002 0：15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java ...
> kill -9 7564

在Windows上使用：

> wmic process get processid,caption,commandline | find "java.exe" | find "server-1.properties"
java.exe    java  -Xmx1G -Xms1G -server -XX:+UseG1GC ... build\libs\kafka_2.10-0.10.1.0.jar"  kafka.Kafka config\server-1.properties    644
> taskkill /pid 644 /f

领导已切换到其中一个从属节点，节点1不再处于同步副本集中：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 1,2,0    Isr: 2,0

但是消息仍然可用于消费，即使采取写入的领导最初是下来：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C

步骤7：使用Kafka Connect导入/导出数据

从控制台写入数据并将其写回控制台是一个方便的起点，但您可能需要使用来自其他来源的数据或将数据从Kafka导出到其他系统。对于许多系统，您可以使用Kafka Connect来导入或导出数据，而不是编写自定义集成代码。

Kafka Connect是Kafka包含的一个工具，用于向Kafka导入和导出数据。它是运行一个可扩展的工具 连接器，其实现的定制逻辑用于与外部系统交互。在这个快速入门中，我们将看到如何使用简单的连接器来运行Kafka Connect，这些连接器将数据从文件导入Kafka主题，并将数据从Kafka主题导出到文件。

首先，我们将从创建一些种子数据开始测试：

> echo -e "foo\nbar" > test.txt

接下来，我们将开始两个连接器中运行的独立模式，这意味着他们在一个单一的，本地的，专用的进程中运行。我们提供三个配置文件作为参数。第一个是Kafka Connect过程的配置，包含常见的配置，如要连接的Kafka代理和数据的序列化格式。其余的配置文件均指定要创建的连接器。这些文件包括唯一的连接器名称，要实例化的连接器类以及连接器所需的任何其他配置。

> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

Kafka包含的这些示例配置文件使用您之前启动的默认本地群集配置，并创建两个连接器：第一个是源连接器，从输入文件读取行并生成每个Kafka主题，第二个是宿连接器它从Kafka主题读取消息，并将其作为输出文件中的一行生成。

在启动期间，您将看到一些日志消息，包括一些指示正在实例化连接器。一旦连接卡夫卡过程已经开始，源连接器应该开始读取行

test.txt

，并将其生产的话题

connect-test

，和水槽连接器应该开始从主题读取消息

connect-test

，并将其写入文件

test.sink.txt

。我们可以通过检查输出文件的内容来验证数据是否已通过整个流水线传送：

>cat test.sink.txt
foo
bar

注意，该数据被存储在卡夫卡主题中

connect-test

，所以我们也可以执行控制台消费者看到主题中的数据（或使用定制消费者的代码来处理它）：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
...

连接器继续处理数据，因此我们可以将数据添加到文件，并查看它通过管道移动：

> echo "Another line" >> test.txt

您应该看到该行出现在控制台使用者输出和sink文件中。

步骤8：使用Kafka Streams来处理数据

Kafka Streams是Kafka的客户端库，用于实时流处理和分析存储在Kafka代理中的数据。这个快速入门示例将演示如何运行在此库中编码的流应用程序。这里的要点

WordCountDemo

示例代码（转换为使用，方便阅读的Java 8 lambda表达式）。

KTable wordCounts = textLines
    //将每个文本行（以空格分隔）转换为单词。
    .flatMapValues（value  - > Arrays.asList（value.toLowerCase（）。split（“\\ W +”）））

    //确保这些字可用作下一个聚合操作的记录键。
    .map（（key，value） - > new KeyValue <>（value，value））

    //计算每个字（记录键）的出现次数，并将结果存储到名为“Counts”的表中。
    .countByKey（“Counts”）

它实现WordCount算法，它从输入文本计算字出现直方图。然而，与其它WORDCOUNT例子，你可能已经看到在此之前，上界数据进行操作时，WORDCOUNT演示应用程序的行为稍有不同，因为它的目的是在一个操作无限的，无限的流数据。与有界变量类似，它是一种有状态算法，用于跟踪和更新单词的计数。然而，由于它必须假定潜在的无界输入数据，它将周期性地输出其当前状态和结果，同时继续处理更多的数据，因为它不知道它何时处理了“全部”输入数据。

我们现在将准备输入数据到Kafka主题，随后将由Kafka Streams应用程序处理。

> echo -e "all streams lead to kafka\nhello kafka streams\njoin kafka summit" > file-input.txt

或在Windows上：

> echo all streams lead to kafka> file-input.txt
> echo hello kafka streams>> file-input.txt
> echo|set /p=join kafka summit>> file-input.txt

接下来，我们这个输入的数据发送到输入主题命名流文件输入使用控制台生产者（在实践中，流数据将可能是连续流入卡夫卡其中应用程序将是启动并运行）：

> bin/kafka-topics.sh --create \
            --zookeeper localhost:2181 \
            --replication-factor 1 \
            --partitions 1 \
            --topic streams-file-input

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-file-input < file-input.txt

我们现在可以运行WordCount演示应用程序来处理输入数据：

> bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo

不会有任何STDOUT输出除日志条目，结果不断地写回另一个主题命名为流-单词计数输出卡夫卡。演示将运行几秒钟，然后，不像典型的流处理应用程序，自动终止。

我们现在可以通过从其输出主题中读取来检查WordCount演示应用程序的输出：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
            --topic streams-wordcount-output \
            --from-beginning \
            --formatter kafka.tools.DefaultMessageFormatter \
            --property print.key=true \
            --property print.value=true \
            --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
            --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

并将以下输出数据打印到控制台：

all     1
lead    1
to      1
hello   1
streams 2
join    1
kafka   3
summit  1

这里，第一列是卡夫卡消息密钥，和第二列是消息值，无论在

java.lang.String

格式。注意，输出实际上是连续的更新流，其中每个数据记录（即，上面原始输出中的每一行）是单个字（也称为记录密钥，例如“kafka”）的更新计数。对于具有相同键的多个记录，每个稍后的记录是前一个记录的更新。

现在，你可以写更多的输入信息到数据流文件输入主题，并观察加入额外的信息流，单词计数输出的话题，反映了更新的字数（例如，使用上述的控制台生产者和消费者的控制台）。

您可以停止通过控制台消费者**按Ctrl-C **。

1.4生态系统

在主分配之外有很多工具与Kafka集成。的生态系统页列出了许多的这些，包括数据流处理系统，Hadoop的集成，监控和部署工具。

1.5从以前的版本升级

从0.8.x，0.9.x或0.10.0.X升级到0.10.1.0

0.10.1.0有线协议更改。通过遵循以下建议的滚动升级计划，您可以确保在升级期间不会停机。但是，请注意在0.10.1.0潜在的重大更改升级之前。注意：由于引入了新协议，因此在升级客户端之前升级Kafka集群非常重要（即0.10.1.x客户端仅支持0.10.1.x或更高版本的代理，而0.10.1.x代理也支持旧客户端）。

滚动升级：

更新所有broker上的server.properties文件，并添加以下属性： - inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2.0,0.9.0.0或0.10.0.0）。- log.message.format.version = CURRENT_KAFKA_VERSION（参见升级后潜在的性能影响对这个配置做什么的详细信息。）
每次升级一个broker：关闭broker，更新代码，然后重新启动它。
一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.10.1.0来覆盖协议版本。
如果先前的消息格式为0.10.0，请将log.message.format.version更改为0.10.1（这是一个无操作，因为消息格式对于0.10.0和0.10.1都相同）。如果您之前的邮件格式版本低于0.10.0，请不要更改log.message.format.version - 此参数只有在所有用户升级到0.10.0.0或更高版本后才会更改。
重新启动代理，使新协议版本生效。
如果此时log.message.format.version仍低于0.10.0，请等待所有用户升级到0.10.0或更高版本，然后将log.message.format.version更改为每个代理上的0.10.1，重新启动它们一个一个。

注意：如果你愿意接受的停机时间，你可以简单地把所有的broker停下来，更新代码，并启动所有的broker。他们将默认从新协议开始。

注：修改协议议版本并且可以做到broker进行升级之后的任何时候重新启动。它不必马上就重启。

潜在重大改变在0.10.1.0

日志保留时间不再基于日志段的上次修改时间。相反，它将基于日志段中的消息的最大时间戳。
日志滚动时间不再取决于日志段创建时间。而是现在基于消息中的时间戳。进一步来说。如果段中第一个消息的时间戳为T，则当新消息的时间戳大于或等于T + log.roll.ms时，日志将被推出
0.10.0的打开文件处理程序将增加约33％，因为为每个段添加时间索引文件。
时间索引和偏移索引共享相同的索引大小配置。因为每个索引条目是偏移索引条目的大小的1.5倍。用户可能需要增加log.index.size.max.bytes以避免潜在的频繁日志滚动。
由于索引文件数量的增加，在一些具有大量日志段（例如> 15K）的代理上，代理启动期间的日志加载过程可能更长。根据我们的实验，将num.recovery.threads.per.data.dir设置为1可以减少日志加载时间。

在0.10.1.0的显著变化

新的Java消费者不再处于测试阶段，我们建议将其用于所有新开发。旧的Scala使用者仍然受支持，但在下一个版本中将被弃用，并将在未来的主要版本中删除。
将--new-consumer/ --new.consumer开关不再需要使用像MirrorMaker和控制台与消费者新的消费工具; 只需要通过一个Kafka代理连接到而不是ZooKeeper集合。此外，控制台消费者与旧消费者的使用已被弃用，并且将在未来的主要版本中删除。
Kafka集群现在可以通过集群ID唯一标识。它将在代理升级到0.10.1.0时自动生成。集群ID可通过kafka.server：type = KafkaServer，name = ClusterId度量标准获得，并且它是元数据响应的一部分。串行器，客户端拦截器和度量记录器可以通过实现ClusterResourceListener接口来接收集群ID。
BrokerState“RunningAsController”（值4）已删除。由于一个错误，代理只会在转换出来之前处于这种状态，因此删除的影响应该是最小的。检测给定代理是否是控制器的推荐方法是通过kafka.controller：type = KafkaController，name = ActiveControllerCount指标。
新的Java消费者现在允许用户通过分区上的时间戳搜索偏移量。
新的Java消费者现在支持从后台线程的心跳。有一个新的配置 max.poll.interval.ms，它控制轮询调用之前，消费者会主动离开组（5分钟默认情况下）之间的最大时间。配置的值 request.timeout.ms必须始终大于max.poll.interval.ms，因为这是一个JoinGroup请求可以在服务器上阻塞而消费是平衡的最大时间，因此，我们已经改变其缺省值，以略高于5分钟。最后，的默认值session.timeout.ms已被调整至10秒，的默认值max.poll.records被改变为500。
当使用授权者和用户没有描述在题目授权，则代理将不再返回TOPIC_AUTHORIZATION_FAILED错误的请求，因为这泄漏的主题名称。相反，将返回UNKNOWN_TOPIC_OR_PARTITION错误代码。当使用生产者和消费者时，这可能导致意外的超时或延迟，因为Kafka客户端通常将在未知的主题错误时自动重试。如果您怀疑这可能发生，您应该咨询客户端日志。
提取响应在默认情况下具有大小限制（对于消费者，为50 MB，对于复制为10 MB）。现有的每个分区限制也适用（对于消费者和复制，为1 MB）。注意，这些限制都不是绝对最大值，如下一点所述。
如果找到大于响应/分区大小限制的消息，则消费者和副本可以进行。更具体地，如果提取的第一非空分区中的第一消息大于任一或两个限制，则仍将返回消息。
重载的构造函数中添加kafka.api.FetchRequest并kafka.javaapi.FetchRequest允许调用者指定分区的顺序（因为顺序是V3显著）。先前存在的构造函数已被弃用，并且在发送请求之前对分区进行重排以避免饥饿问题。

新协议版本

ListOffsetRequest v1支持基于时间戳的精确偏移量搜索。
MetadataResponse v2引入了一个新字段：“cluster_id”。
FetchRequest v3支持限制响应大小（除了现有的每个分区限制），它返回大于限制的消息，如果需要进行，并且请求中的分区的顺序现在是重要的。
JoinGroup v1引入了一个新字段：“rebalance_timeout”。

从0.8.x或0.9.x升级到0.10.0.0

0.10.0.0具有潜在的重大更改（请在升级前检查），并可能升级后性能的影响。通过遵循以下建议的滚动升级计划，您可以确保在升级期间和之后不会出现停机时间和性能影响。注意：由于引入了新协议，因此在升级客户端之前升级Kafka集群非常重要。

注释客户提供0.9.0.0版本：由于0.9.0.0引入了一个错误，即依赖于ZooKeeper的客户（老斯卡拉高层次消费者和MirrorMaker如果与老消费者使用）不会与0.10.0.xbroker的工作。因此，0.9.0.0客户应升级到0.9.0.1 之前的broker都升级到0.10.0.x. 对于0.8.X或0.9.0.1客户端，此步骤不是必需的。

滚动升级：

更新所有代理上的server.properties文件，并添加以下属性： - inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2或0.9.0.0）。- log.message.format.version = CURRENT_KAFKA_VERSION（参见升级后潜在的性能影响对这个配置做什么的详细信息。）
升级经纪商。这可以通过简单地把它放下，更新代码，并重新启动代理一次完成。
一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.10.0.0来颠覆协议版本。注意：您不应该触摸log.message.format.version - 此参数应该只有更改一旦所有用户已升级到0.10.0.0
重新启动代理，使新协议版本生效。
一旦所有消费者已升级到0.10.0，将每个代理上的log.message.format.version更改为0.10.0，然后逐个重新启动它们。

注意：如果你愿意接受的停机时间，你可以简单地把所有的broker下来，更新代码，并开始所有的人。他们将默认从新协议开始。

注：碰碰协议版本并重新启动可以做到任何时候broker进行了升级之后。它不必马上就在。

升级到0.10.0.0后潜在的性能影响

0.10.0中的消息格式包括新的时间戳字段，并使用压缩消息的相对偏移量。磁盘消息格式可以通过server.properties文件中的log.message.format.version进行配置。默认磁盘消息格式为0.10.0。如果客户端客户端的版本低于0.10.0.0，它只能理解0.10.0之前的邮件格式。在这种情况下，代理能够在将响应发送到旧版本上的消费者之前将消息从0.10.0格式转换为较早的格式。但是，在这种情况下，代理不能使用零拷贝传输。Kafka社区关于性能影响的报告显示，在升级后，CPU利用率从20％提高到100％，这迫使所有客户端立即升级，使性能恢复正常。为了避免在消费者升级到0.10.0.0之前进行此类消息转换，可以在将代理升级到0.10.0.0时将log.message.format.version设置为0.8.2或0.9.0。这样，代理仍然可以使用零拷贝传输将数据发送给旧的消费者。一旦消费者升级，就可以在代理上将消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持转换以确保兼容性，并且可以用于支持尚未更新到较新客户端的一些应用，但是即使在过度配置的群集上也支持所有消费者流量是不切实际的。因此，当代理升级但绝大多数客户没有升级时，尽可能避免消息转换是至关重要的。

对于升级到0.10.0.0的客户端，不会对性能产生影响。

注：通过设置消息格式版本，一是证明现有的所有信息都在或低于该消息格式版本。否则消费者在0.10.0.0之前可能会中断。特别是，在消息格式设置为0.10.0之后，不应将其更改回较早的格式，因为它可能会在0.10.0.0之前的版本上中断用户。

注：由于每个消息中引入的附加时间戳，生产者发送小消息可能会看到，因为增加开销的消息吞吐量下降。同样，复制现在每个消息传输额外的8个字节。如果您运行的集群的网络容量接近，可能会淹没网卡，并看到由于过载的故障和性能问题。

注意：如果已启用压缩对生产者，您可能会注意到降低生产吞吐量和/或降低压缩比在某些情况下，broker。当接收压缩消息时，0.10.0代理避免重新压缩消息，这通常减少延迟并提高吞吐量。然而，在某些情况下，这可以减少生产商的批量化尺寸，这可能导致较差的吞吐量。如果发生这种情况，用户可以调整生产者的linger.ms和batch.size以获得更好的吞吐量。此外，用于压缩具有snappy的消息的生成器缓冲器小于代理使用的生成器缓冲器，这可能对磁盘上的消息的压缩比有负面影响。我们打算在未来的Kafka版本中进行配置。

潜在断裂变化0.10.0.0

从Kafka 0.10.0.0开始，Kafka中的消息格式版本表示为Kafka版本。例如，消息格式0.9.0指的是Kafka 0.9.0支持的最高消息版本。
已经介绍了消息格式0.10.0，并且默认使用它。它包括消息中的时间戳字段和相对偏移量用于压缩消息。
已经引入了ProduceRequest / Response v2，它默认使用支持消息格式0.10.0
FetchRequest / Response v2已经被引入，它默认使用支持消息格式0.10.0
MessageFormatter接口从改变def writeTo(key: Array[Byte], value: Array[Byte], output: PrintStream)到 def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]], output: PrintStream)
MessageReader接口从改变def readMessage(): KeyedMessage[Array[Byte], Array[Byte]]到 def readMessage(): ProducerRecord[Array[Byte], Array[Byte]]
MessageFormatter的包从改变kafka.tools到kafka.common
MessageReader的包从改变kafka.tools到kafka.common
MirrorMakerMessageHandler不再公开handle(record: MessageAndMetadata[Array[Byte], Array[Byte]])，因为它从未被调用的方法。
0.7 KafkaMigrationTool不再与Kafka一起打包。如果您需要从0.7迁移到0.10.0，请先迁移到0.8，然后按照记录的升级过程从0.8升级到0.10.0。
新的消费已经标准化，它的API接受java.util.Collection作为方法参数序列类型。可能必须更新现有代码以使用0.10.0客户端库。
LZ4压缩消息处理已更改为使用可互操作的成帧规范（LZ4f v1.5.1）。为了保持与旧客户端的兼容性，此更改仅适用于消息格式0.10.0和更高版本。使用v0 / v1（消息格式0.9.0）生成/获取LZ4压缩消息的客户端应继续使用0.9.0成帧实现。使用Produce / Fetch协议v2或更高版本的客户端应使用可互操作的LZ4f成帧。可互操作的LZ4库的列表可在http://www.lz4.org/

在0.10.0.0的显着变化

从卡夫卡0.10.0.0开始，任命了新的客户端库卡夫卡流可用于流处理存储在卡夫卡的主题数据。这个新的客户端库仅适用于0.10.x和向上版本的代理，由于上面提到的消息格式更改。欲了解更多信息，请阅读本节。
配置参数的默认值receive.buffer.bytes现在是64K的新的消费。
新的消费现在公开的配置参数exclude.internal.topics，从意外被列入正则表达式订阅限制内部主题（如消费者偏移主题）。默认情况下，启用。
旧的Scala生产者已被弃用。用户应尽快将其代码迁移到kafka-clients JAR中包括的Java生成器。
新的消费者API已标记为稳定。

从0.8.0，0.8.1.X或0.8.2.X升级到0.9.0.0

0.9.0.0具有潜在的重大更改（请在升级前检查），并从以前的版本中，券商间的协议更改。这意味着升级的代理和客户端可能与旧版本不兼容。在升级客户端之前升级Kafka集群很重要。如果您使用MirrorMaker下游集群，则应首先升级。

滚动升级：

更新所有代理上的server.properties文件，并添加以下属性：inter.broker.protocol.version = 0.8.2.X
升级经纪商。这可以通过简单地把它放下，更新代码，并重新启动代理一次完成。
一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.9.0.0来颠覆协议版本。
重新启动代理，使新协议版本生效

注意：如果你愿意接受的停机时间，你可以简单地把所有的broker下来，更新代码，并开始所有的人。他们将默认从新协议开始。

注：碰碰协议版本并重新启动可以做到任何时候broker进行了升级之后。它不必马上就在。

潜在断裂变化在0.9.0.0

不再支持Java 1.6。
不再支持Scala 2.9。
默认情况下，1000以上的代理ID保留为自动分配的代理ID。如果您的群集具有高于该阈值的现有代理ID，请确保相应地增加reserved.broker.max.id代理配置属性。
配置参数replica.lag.max.messages已删除。分区领导在决定哪些副本处于同步时将不再考虑滞后消息的数量。
配置参数replica.lag.time.max.ms现在不仅指自上次从副本获取请求后经过的时间，而且还指自副本上次被捕获以来的时间。仍然从领导者抓取邮件但未赶上replica.lag.time.max.ms中的最新邮件的副本将被认为不同步。
压缩的主题不再接受没有键的消息，如果尝试这种情况，生产者抛出异常。在0.8.x中，没有键的消息将导致日志压缩线程随后抱怨和退出（并停止压缩所有压缩的主题）。
MirrorMaker不再支持多个目标集群。因此，它只接受一个--consumer.config参数。要镜像多个源集群，每个源集群至少需要一个MirrorMaker实例，每个源集群都有自己的使用者配置。
下包装工具*org.apache.kafka.clients.tools 已移至*org.apache.kafka.tools。。所有包含的脚本仍将照常工作，只有直接导入这些类的自定义代码将受到影响。
在kafka-run-class.sh中更改了默认的Kafka JVM性能选项（KAFKA_JVM_PERFORMANCE_OPTS）。
kafka-topics.sh脚本（kafka.admin.TopicCommand）现在退出，失败时出现非零退出代码。
kafka-topics.sh脚本（kafka.admin.TopicCommand）现在将在主题名称由于使用“。”而导致风险度量标准冲突时打印警告。或“_”，以及在实际碰撞的情况下的错误。
kafka-console-producer.sh脚本（kafka.tools.ConsoleProducer）将使用Java生成器而不是旧的Scala生产者作为默认值，用户必须指定“old-producer”才能使用旧生产者。
默认情况下，所有命令行工具都会将所有日志消息打印到stderr而不是stdout。

0.9.0.1中的显着变化

可以通过将broker.id.generation.enable设置为false来禁用新的代理标识生成功能。
默认情况下，配置参数log.cleaner.enable现在为true。这意味着具有cleanup.policy = compact的主题现在将默认压缩，并且128 MB的堆将通过log.cleaner.dedupe.buffer.size分配给更清洁的进程。您可能需要根据您对压缩主题的使用情况，查看log.cleaner.dedupe.buffer.size和其他log.cleaner配置值。
默认情况下，新使用者的配置参数fetch.min.bytes的默认值为1。

折旧在0.9.0.0

已弃用从kafka-topics.sh脚本（kafka.admin.TopicCommand）更改主题配置。今后，请使用kafka-configs.sh脚本（kafka.admin.ConfigCommand）来实现此功能。
kafka-consumer-offset-checker.sh（kafka.tools.ConsumerOffsetChecker）已被弃用。今后，请使用kafka-consumer-groups.sh（kafka.admin.ConsumerGroupCommand）执行此功能。
kafka.tools.ProducerPerformance类已被弃用。接下来，请为此功能使用org.apache.kafka.tools.ProducerPerformance（kafka-producer-perf-test.sh也将更改为使用新类）。
生成器config block.on.buffer.full已被弃用，并将在以后的版本中删除。目前，其默认值已更改为false。KafkaProducer将不再抛出BufferExhaustedException，而是使用max.block.ms值来阻止，之后会抛出一个TimeoutException异常。如果block.on.buffer.full属性显式设置为true，它将设置max.block.ms为Long.MAX_VALUE和metadata.fetch.timeout.ms将不被执行

从0.8.1升级到0.8.2

0.8.2与0.8.1完全兼容。通过简单地将其降级??，更新代码并重新启动，可以一次完成一个代理的升级。

从0.8.0升级到0.8.1

0.8.1完全兼容0.8。通过简单地将其降级??，更新代码并重新启动，可以一次完成一个代理的升级。

升级从0.7

版本0.7与较新版本不兼容。对API，ZooKeeper数据结构，协议和配置进行了主要更改，以便添加复制（在0.7中缺失）。从0.7到更高版本的升级需要特殊的工具进行迁移。此迁移可以在不停机的情况下完成。

2. API

Kafka包括四个核心apis：

生产者API允许应用程序发送数据流的卡夫卡集群中的主题。
消费者 API允许应用程序从卡夫卡集群中的主题读取数据流。
流 API允许将来自输入题目数据流输出主题。
连接 API允许实现连接器，不断从某些源系统或应用程序拉进卡夫卡或卡夫卡推入一些水槽系统或应用程序。

Kafka公开了其所有的功能超过一个语言独立的协议，客户端可以在许多编程语言。但是，只有Java客户端作为主要Kafka项目的一部分进行维护，其他Java客户端作为独立的开源项目提供。非Java客户的名单可在这里。

2.1生产者API

Producer API允许应用程序向Kafka集群中的主题发送数据流。

示出了如何使用生产者实施例中给出的javadocs。

要使用生成器，可以使用以下maven依赖关系：

<dependency>
            <groupId> org.apache.kafka </ groupId>
            <artifactId> kafka-clients </ artifactId>
            <version> 0.10.1.0 </ version>
        </ dependency>

2.2消费者API

Consumer API允许应用程序从Kafka集群中的主题中读取数据流。

示出了如何使用消费者的例子中给出的javadocs。

要使用消费者，可以使用以下maven依赖关系：

<dependency>
            <groupId> org.apache.kafka </ groupId>
            <artifactId> kafka-clients </ artifactId>
            <version> 0.10.1.0 </ version>
        </ dependency>

2.3 Streams API

该流 API允许将来自输入题目数据流输出主题。

举例说明如何使用这个库中给出的javadoc

在使用流API的其他文档，请点击这里。

要使用Kafka Streams，您可以使用以下maven依赖关系：

<dependency>
            <groupId> org.apache.kafka </ groupId>
            <artifactId> kafka-streams </ artifactId>
            <version> 0.10.1.0 </ version>
        </ dependency>

2.4连接API

Connect API允许实现连续地从一些源数据系统拉入Kafka或从Kafka推送到某个sink数据系统的连接器。

Connect的许多用户不需要直接使用此API，但是他们可以使用预构建的连接器，而无需编写任何代码。使用连接的更多信息，请点击这里。

谁想要实现自定义的接口可以看到的javadoc。

2.5遗留API

Kafka还包括一个更有限的传统生产者和消费者api。这些旧的Scala API已弃用，并且仍然可用于兼容性目的。对他们的信息可以在这里找到这里。

3.配置

卡夫卡在使用键值对的属性文件格式进行配置。这些值可以从文件或编程方式提供。

3.1Broker配置

基本配置如下：

broker.id
log.dirs
zookeeper.connect

主题级配置，缺省值为在更详细地讨论如下。

名称

描述

类型

默认

有效值

重要性

zookeeper.connect

Zookeeper主机字符串

串

高

advertised.host.name

DEPRECATED：仅当未设置advertised.listeners'或listeners'时使用。请改用advertised.listeners。要发布到ZooKeeper以供客户端使用的主机名。在IaaS环境中，这可能需要与代理绑定的接口不同。如果这没有设置，它将使用host.name的值（如果配置）。否则它将使用从java.net.InetAddress.getCanonicalHostName（）返回的值。

串

空值

高

advertised.listeners

发布到ZooKeeper的客户端使用的侦听器，如果不同于上面的侦听器。在IaaS环境中，这可能需要与代理绑定的接口不同。如果没有设置，将使用listeners的值。

串

空值

高

advertised.port

DEPRECATED：仅当未设置advertised.listeners'或listeners'时使用。请改用advertised.listeners。发布到ZooKeeper的端口供客户端使用。在IaaS环境中，这可能需要与代理绑定的端口不同。如果没有设置，它将发布代理绑定到的相同端口。

int

空值

高

auto.create.topics.enable

启用在服务器上自动创建主题

布尔

真正

高

auto.leader.rebalance.enable

启用自动前导平衡。如果需要，后台线程会定期检查并触发领导者平衡

布尔

真正

高

background.threads

用于各种后台处理任务的线程数

int

[1，...]

高

broker.id

此服务器的代理标识。如果未设置，将生成唯一的代理标识。为了避免zookeeper生成的代理标识和用户配置的代理标识之间的冲突，生成的代理idsstart从reserved.broker.max.id + 1。

int

-1

高

compression.type

指定给定主题的最终压缩类型。此配置接受标准压缩编解码器（'gzip'，'snappy'，'lz4'）。它还接受“未压缩”，这相当于没有压缩; 和“生成器”，意味着保留由生产者设置的原始压缩编解码器。

串

生产者

高

delete.topic.enable

启用删除主题。如果此配置已关闭，通过管理工具删除主题将没有任何效果

布尔

假

高

host.name

DEPRECATED：仅当未设置listeners时使用。使用listeners代替。代理的主机名。如果设置，它将只绑定到此地址。如果没有设置，它将绑定到所有接口

串

“”

高

leader.imbalance.check.interval.seconds

控制器触发分区重新平衡检查的频率

长

300

高

leader.imbalance.per.broker.percentage

允许每个broker领导不平衡的比率。如果每个broker的价值超过此价值，控制器将触发领导者余额。该值以百分比形式指定。

int

高

listeners

监听器列表 - 我们将监听的URI的逗号分隔列表及其协议。将hostname指定为0.0.0.0以绑定到所有接口。将主机名留空以绑定到默认接口。合法侦听器列表示例：PLAINTEXT：// myhost：9092，TRACE：//：9091 PLAINTEXT：//0.0.0.0：9092，TRACE：// localhost：9093

串

空值

高

log.dir

保存日志数据的目录（对log.dirs属性的补充）

串

/ tmp / kafka-logs

高

log.dirs

保存日志数据的目录。如果未设置，则使用log.dir中的值

串

空值

高

log.flush.interval.messages

消息刷新到磁盘之前在日志分区上累积的消息数

long

9223372036854775807

[1，...]

高

log.flush.interval.ms

任何主题中的消息在刷新到磁盘之前保存在内存中的最大时间（以毫秒为单位）。如果未设置，则使用log.flush.scheduler.interval.ms中的值

long

空值

高

log.flush.offset.checkpoint.interval.ms

我们更新作为日志恢复点的最后一次冲洗的持久记录的频率

int

60000

[0，...]

高

log.flush.scheduler.interval.ms

日志刷新器检查是否有任何日志需要刷新到磁盘的频率（以毫秒为单位）

long

9223372036854775807

高

log.retention.bytes

删除日志之前的日志的最大大小

long

-1

高

log.retention.hours

删除日志文件之前保留的小时数（以小时为单位），第三级为log.retention.ms属性

int

168

高

log.retention.minutes

在删除日志文件之前保持日志文件的分钟数（以分钟为单位），次于log.retention.ms属性。如果未设置，则使用log.retention.hours中的值

int

空值

高

log.retention.ms

在删除日志文件之前保留日志文件的毫秒数（以毫秒为单位），如果未设置，则使用log.retention.minutes中的值

long

空值

高

log.roll.hours

新日志段推出之前的最长时间（以小时为单位），次于log.roll.ms属性

int

168

[1，...]

高

log.roll.jitter.hours

从logRollTimeMillis（以小时为单位）减去的最大抖动，继承于log.roll.jitter.ms属性

int

[0，...]

高

log.roll.jitter.ms

从logRollTimeMillis中减去的最大抖动（以毫秒为单位）。如果未设置，则使用log.roll.jitter.hours中的值

long

空值

高

log.roll.ms

新日志段推出之前的最长时间（以毫秒为单位）。如果未设置，则使用log.roll.hours中的值

long

空值

高

log.segment.bytes

单个日志文件的最大大小

int

1073741824

[14，...]

高

log.segment.delete.delay.ms

从文件系统中删除文件之前等待的时间

long

60000

[0，...]

高

message.max.bytes

服务器可以接收的消息的最大大小

int

1000012

[0，...]

高

min.insync.replicas

当生产者将ack设置为“all”（或“-1”）时，min.insync.replicas指定必须确认写入的副本的最小数量，以使写入被认为成功。如果不能满足这个最小值，那么生产者将引发异常（NotEnoughReplicas或NotEnoughReplicasAfterAppend）。当一起使用时，min.insync.replicas和acks允许你强制更强的耐久性保证。典型的情况是创建一个复制因子为3的主题，将min.insync.replicas设置为2，并产生一个“all”的acks。这将确保生成器在大多数副本没有接收到写入时引发异常。

int

[1，...]

高

num.io.threads

服务器用于执行网络请求的io线程数

int

[1，...]

高

num.network.threads

服务器用于处理网络请求的网络线程数

int

[1，...]

高

num.recovery.threads.per.data.dir

每个数据目录的线程数，用于在启动时进行日志恢复并在关闭时刷新

int

[1，...]

高

num.replica.fetchers

用于从源代理复制消息的提取线程数。增加此值可以提高跟随器代理中的I / O并行度。

int

高

offset.metadata.max.bytes

与偏移提交关联的元数据条目的最大大小

int

4096

高

offsets.commit.required.acks

可以接受提交之前所需的ack。通常，不应覆盖默认值（-1）

短

-1

高

offsets.commit.timeout.ms

偏移提交将被延迟，直到偏移主题的所有副本都收到提交或达到此超时。这类似于生产者请求超时。

int

5000

[1，...]

高

offsets.load.buffer.size

用于在将偏移量装入缓存时从偏移段读取的批量大小。

int

5242880

[1，...]

高

offsets.retention.check.interval.ms

检查陈旧偏移的频率

long

600000

[1，...]

高

offsets.retention.minutes

偏移主题的日志保留时间（分钟）

int

1440

[1，...]

高

offsets.topic.compression.codec

用于偏移主题的压缩编解码器 - 压缩可以用于实现“原子”提交

int

高

offsets.topic.num.partitions

偏移提交主题的分区数（部署后不应更改）

int

[1，...]

高

offsets.topic.replication.factor

偏移主题的复制因子（设置较高以确保可用性）。为了确保偏移主题的有效复制因子是配置的值，活动代理的数量必须至少是对offsetets主题的第一次请求时的复制因子。如果不是，偏移主题创建将失败或它将获得一个复制因子min（alive Brokers，配置的复制因子）

短

[1，...]

高

offsets.topic.segment.bytes

偏移主题段字节应保持相对较小，以便于加快日志压缩和缓存加载

int

104857600

[1，...]

高

port

DEPRECATED：仅当未设置listeners时使用。使用listeners代替。端口监听和接受连接

int

9092

高

queued.max.requests

在阻止网络线程之前允许的排队请求数

int

500

[1，...]

高

quota.consumer.default

DEPRECATED：仅在未配置动态默认配额时使用要么在Zookeeper。由clientId / consumer组区分的任何消费者如果每秒获取的字节数超过此值，则会受到限制

long

9223372036854775807

[1，...]

高

quota.producer.default

DEPRECATED：仅在未配置动态默认配额时使用，要么在Zookeeper。由clientId区分的任何生产者如果每秒产生的字节数大于此值，则会受到限制

long

9223372036854775807

[1，...]

高

replica.fetch.min.bytes

每个获取响应所需的最小字节数。如果没有足够的字节，请等待到replicaMaxWaitTimeMs

int

高

replica.fetch.wait.max.ms

由跟随者副本发出的每个获取器请求的最大等待时间。此值应始终小于replica.lag.time.max.ms以防止低吞吐量主题的ISR频繁收缩

int

500

高

replica.high.watermark.checkpoint.interval.ms

高水位线保存到磁盘的频率

long

5000

高

replica.lag.time.max.ms

如果跟随者没有发送任何获取请求或者没有消耗到领导者日志结束偏移至少这个时间，领导者将从isr去除跟随者

long

10000

高

replica.socket.receive.buffer.bytes

用于网络请求的套接字接收缓冲区

int

65536

高

replica.socket.timeout.ms

网络请求的套接字超时。其值应至少为replica.fetch.wait.max.ms

int

30000

高

request.timeout.ms

配置控制客户端等待请求响应的最大时间量。如果在超时时间之前没有收到响应，客户端将在必要时重新发送请求，如果重试次数耗尽，则请求失败。

int

30000

高

socket.receive.buffer.bytes

套接字服务器套接字的SO_RCVBUF缓冲区。如果值为-1，将使用操作系统默认值。

int

102400

高

socket.request.max.bytes

套接字请求中的最大字节数

int

104857600

[1，...]

高

socket.send.buffer.bytes

套接字服务器套接字的SO_SNDBUF缓冲区。如果值为-1，将使用操作系统默认值。

int

102400

高

unclean.leader.election.enable

指示是否启用不在ISR集中的副本作为最后手段被选为leader，即使这样做可能会导致数据丢失

布尔

真正

高

zookeeper.connection.timeout.ms

客户端等待与zookeeper建立连接的最长时间。如果未设置，则使用zookeeper.session.timeout.ms中的值

int

空值

高

zookeeper.session.timeout.ms

Zookeeper会话超时

int

6000

高

zookeeper.set.acl

设置客户端使用安全ACL

布尔

假

高

broker.id.generation.enable

在服务器上启用自动代理标识生成。启用时，应检查为reserved.broker.max.id配置的值。

布尔

真正

中

broker.rack

代理的机架。这将在机架感知复制分配中用于容错。示例：RACK1，us-east-1d

串

空值

中

connections.max.idle.ms

空闲连接超时：服务器socket处理器线程关闭空闲的连接超过此

long

600000

中

controlled.shutdown.enable

启用服务器的受控关闭

布尔

真正

中

controlled.shutdown.max.retries

受控关机可能由于多种原因而失败。这将确定发生此类故障时的重试次数

int

中

controlled.shutdown.retry.backoff.ms

在每次重试之前，系统需要时间从导致先前故障的状态（控制器故障切换，副本滞后等）恢复。此配置确定重试之前等待的时间量。

long

5000

中

controller.socket.timeout.ms

控制器到代理通道的套接字超时

int

30000

中

default.replication.factor

默认复制因子为自动创建的主题

int

中

fetch.purgatory.purge.interval.requests

提取请求炼狱的清除间隔（请求数）

int

1000

中

group.max.session.timeout.ms

注册用户的最大允许会话超时。更长的超时使消费者有更多的时间在心跳之间处理消息，但花费更长的时间来检测故障。

int

300000

中

group.min.session.timeout.ms

注册用户的最小允许会话超时。较短的超时导致更快的故障检测，代价是更频繁的消费者心跳，这可能压倒broker资源。

int

6000

中

inter.broker.protocol.version

指定将使用代理间协议的哪个版本。这通常是在所有代理升级到新版本后发生。一些有效值的示例是：0.8.0,0.8.1,0.8.1.1,0.8.2,0.8.2.0,0.8.2.1,0.9.0.0,0.9.0.1检查完整列表的ApiVersion。

串

0.10.1-IV2

中

log.cleaner.backoff.ms

当没有日志要清理时，睡眠的时间量

long

15000

[0，...]

中

log.cleaner.dedupe.buffer.size

用于所有清除程序线程的日志重复数据删除的总内存

long

134217728

中

log.cleaner.delete.retention.ms

删除记录保留多长时间？

long

86400000

中

log.cleaner.enable

启用日志清理器进程在服务器上运行？如果使用任何包含cleanup.policy = compact的主题包括内部偏移主题，应该启用。如果禁用，那些主题将不会被压缩并且尺寸不断增大。

布尔

真正

中

log.cleaner.io.buffer.load.factor

日志清除器重复数据删除缓冲区负载因子重复数据删除缓冲区已满的百分比。较高的值将允许同时清除更多的日志，但会导致更多的哈希冲突

双

0.9

中

log.cleaner.io.buffer.size

用于清理所有清除程序线程的日志清除器I / O缓冲区的总内存

int

524288

[0，...]

中

log.cleaner.io.max.bytes.per.second

日志清理器将被节流，以使其读写I / O的总和将平均小于此值

双

1.7976931348623157E308

中

log.cleaner.min.cleanable.ratio

脏日志与日志的总日志的最小比率，以便有资格进行清理

双

0.5

中

log.cleaner.min.compaction.lag.ms

消息在日志中保持未压缩的最小时间。仅适用于正在压缩的日志。

长

中

log.cleaner.threads

用于日志清理的后台线程数

int

[0，...]

中

log.cleanup.policy

超出保留时间段的段的默认清除策略。逗号分隔的有效策略列表。有效的策略是：“delete”和“compact”

列表

[删除]

[紧凑，删除]

中

log.index.interval.bytes

我们向偏移索引添加条目的间隔

int

4096

[0，...]

中

log.index.size.max.bytes

偏移索引的最大大小（以字节为单位）

int

10485760

[4，...]

中

log.message.format.version

指定代理将用于将消息附加到日志的消息格式版本。该值应为有效的ApiVersion。一些示例是：0.8.2,0.9.0.0,0.10.0，检查ApiVersion以获取更多详细信息。通过设置特定的消息格式版本，用户证明磁盘上的所有现有消息小于或等于指定的版本。不正确地设置此值将导致旧版本的客户中断，因为他们将接收到他们不明白的格式的邮件。

串

0.10.1-IV2

中

log.message.timestamp.difference.max.ms

代理接收消息时的时间戳和消息中指定的时间戳之间允许的最大差异。如果log.message.timestamp.type = CreateTime，则如果时间戳的差异超过此阈值，则会拒绝邮件。如果log.message.timestamp.type = LogAppendTime，则忽略此配置。

long

9223372036854775807

[0，...]

中

log.message.timestamp.type

定义消息中的时间戳是消息创建时间还是日志附加时间。该值应为CreateTime或LogAppendTime

串

CreateTime

[CreateTime，LogAppendTime]

中

log.preallocate

应该在创建新段时预分配文件？如果您在Windows上使用Kafka，则可能需要将其设置为true。

布尔

假

中

log.retention.check.interval.ms

日志清除程序检查任何日志是否有资格删除的频率（以毫秒为单位）

long

300000

[1，...]

中

max.connections.per.ip

我们从每个IP地址允许的最大连接数

int

2147483647

[1，...]

中

max.connections.per.ip.overrides

per-ip或hostname覆盖默认最大连接数

串

“”

中

num.partitions

每个主题的默认日志分区数

int

[1，...]

中

principal.builder.class

实现PrincipalBuilder接口的类的完全限定名，该接口当前用于构建与SSL SecurityProtocol的连接的Principal。

类

class org.apache.kafka.common.security.auth.DefaultPrincipalBuilder

中

producer.purgatory.purge.interval.requests

生产者请求炼狱的清除间隔（请求数）

int

1000

中

replica.fetch.backoff.ms

发生抓取分区错误时休眠的时间。

int

1000

[0，...]

中

replica.fetch.max.bytes

要为每个分区获取的消息的字节数。这不是绝对最大值，如果提取的第一个非空分区中的第一个消息大于此值，则仍会返回消息以确保可以进行进度。由broker接受的最大邮件大小是通过定义message.max.bytes（broker配置）或max.message.bytes（主题配置）。

int

1048576

[0，...]

中

replica.fetch.response.max.bytes

整个获取响应所需的最大字节数。这不是绝对最大值，如果提取的第一个非空分区中的第一个消息大于此值，则仍会返回消息以确保可以进行进度。由broker接受的最大邮件大小是通过定义message.max.bytes（broker配置）或max.message.bytes（主题配置）。

int

10485760

[0，...]

中

reserved.broker.max.id

可用于broker.id的最大数量

int

1000

[0，...]

中

sasl.enabled.mechanisms

在Kafka服务器中启用的SASL机制的列表。该列表可以包含安全提供者可用的任何机制。默认情况下仅启用GSSAPI。

列表

[GSSAPI]

中

sasl.kerberos.kinit.cmd

Kerberos kinit命令路径。

串

/ usr / bin / kinit

中

sasl.kerberos.min.time.before.relogin

登录线程在刷新尝试之间的休眠时间。

long

60000

中

sasl.kerberos.principal.to.local.rules

用于从主体名称到短名称（通常是操作系统用户名）的映射规则列表。按顺序评估规则，并且使用与主体名称匹配的第一个规则将其映射到短名称。将忽略列表中的任何后续规则。默认情况下，{username} / {hostname} @ {REALM}形式的主体名称映射到{username}。有关格式的详细信息，请参阅安全授权和访问控制列表。

列表

[默认]

中

sasl.kerberos.service.name

Kafka运行的Kerberos主体名称。这可以在Kafka的JAAS配置或Kafka的配置中定义。

串

空值

中

sasl.kerberos.ticket.renew.jitter

添加到更新时间的随机抖动的百分比。

double

0.05

中

sasl.kerberos.ticket.renew.window.factor

登录线程将睡眠，直到从上次刷新到票据到期的时间的指定窗口因子已经到达，在该时间它将尝试更新票证。

double

0.8

中

sasl.mechanism.inter.broker.protocol

SASL机制用于代理间通信。默认值为GSSAPI。

串

GSSAPI

中

security.inter.broker.protocol

用于在代理之间通信的安全协议。有效值为：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL。

串

纯文本

中

ssl.ciphersuites

密码套件列表。这是一种命名的认证，加密，MAC和密钥交换算法的组合，用于使用TLS或SSL网络协议协商网络连接的安全设置。默认情况下，支持所有可用的密码套件。

列表

空值

中

Ssl.client.auth

配置kafka代理以请求客户端认证。以下设置是常见的： ssl.client.auth=required 如果设置为必需，则需要客户端认证。 ssl.client.auth=requested这意味着客户端认证是可选的。不同于请求，如果设置此选项，客户端可以选择不提供有关其自身的身份验证信息 ssl.client.auth=none 这意味着不需要客户端身份验证。

串

没有

[必填，请求，无]

中

ssl.enabled.protocols

为SSL连接启用的协议列表。

列表

[TLSv1.2，TLSv1.1，TLSv1]

中

ssl.key.password

密钥存储文件中的私钥的密码。这对于客户端是可选的。

密码

空值

中

ssl.keymanager.algorithm

密钥管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。

串

SunX509

中

ssl.keystore.location

密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。

串

空值

中

ssl.keystore.password

密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。

密码

空值

中

ssl.keystore.type

密钥存储文件的文件格式。这对于客户端是可选的。

串

JKS

中

ssl.protocol

用于生成SSLContext的SSL协议。默认设置为TLS，这在大多数情况下是正常的。最近的JVM中允许的值为TLS，TLSv1.1和TLSv1.2。在较早的JVM中可能支持SSL，SSLv2和SSLv3，但是由于已知的安全漏洞，它们的使用不受欢迎。

串

TLS

中

ssl.provider

用于SSL连接的安全提供程序的名称。默认值是JVM的默认安全提供程序。

串

空值

中

ssl.trustmanager.algorithm

信任管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的信任管理器工厂算法。

串

PKIX

中

ssl.truststore.location

信任存储文件的位置。

串

空值

中

ssl.truststore.password

信任存储文件的密码。

密码

空值

中

ssl.truststore.type

信任存储文件的文件格式。

串

JKS

中

authorizer.class.name

应该用于授权的授权程序类

串

“”

低

metric.reporters

用作度量报告器的类的列表。实现MetricReporter接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。

列表

[]

低

metrics.num.samples

维持计算度量的样本数。

int

[1，...]

低

metrics.sample.window.ms

计算度量样本的时间窗口。

long

30000

[1，...]

低

quota.window.num

要在客户端配额的内存中保留的样本数

int

[1，...]

低

quota.window.size.seconds

客户端配额的每个样本的时间跨度

int

[1，...]

低

replication.quota.window.num

要在复制配额的内存中保留的样本数

int

[1，...]

低

replication.quota.window.size.seconds

复制配额的每个样本的时间跨度

int

[1，...]

低

ssl.endpoint.identification.algorithm

端点标识算法，使用服务器证书验证服务器主机名。

串

空值

低

ssl.secure.random.implementation

用于SSL加密操作的SecureRandom PRNG实现。

串

空值

低

zookeeper.sync.time.ms

ZK领导者背后有多远ZK追随者

int

2000年

低

ssl.client.auth=required 如果设置为必需，则需要客户端认证。
ssl.client.auth=requested这意味着客户端认证是可选的。不同于请求，如果设置此选项，客户端可以选择不提供有关其自身的身份验证信息
ssl.client.auth=none 这意味着不需要客户端身份验证。

串没有 [必填，请求，无] 中 ssl.enabled.protocols 为SSL连接启用的协议列表。列表 [TLSv1.2，TLSv1.1，TLSv1] 中 ssl.key.password 密钥存储文件中的私钥的密码。这对于客户端是可选的。密码空值中 ssl.keymanager.algorithm 密钥管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。串 SunX509 中 ssl.keystore.location 密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。串空值中 ssl.keystore.password 密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。密码空值中 ssl.keystore.type 密钥存储文件的文件格式。这对于客户端是可选的。串 JKS 中 ssl.protocol 用于生成SSLContext的SSL协议。默认设置为TLS，这在大多数情况下是正常的。最近的JVM中允许的值为TLS，TLSv1.1和TLSv1.2。在较早的JVM中可能支持SSL，SSLv2和SSLv3，但是由于已知的安全漏洞，它们的使用不受欢迎。串 TLS 中 ssl.provider 用于SSL连接的安全提供程序的名称。默认值是JVM的默认安全提供程序。串空值中 ssl.trustmanager.algorithm 信任管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的信任管理器工厂算法。串 PKIX 中 ssl.truststore.location 信任存储文件的位置。串空值中 ssl.truststore.password 信任存储文件的密码。密码空值中 ssl.truststore.type 信任存储文件的文件格式。串 JKS 中 authorizer.class.name 应该用于授权的授权程序类串 “” 低 metric.reporters 用作度量报告器的类的列表。实现

MetricReporter

接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。列表 [] 低 metrics.num.samples 维持计算度量的样本数。 int 2 [1，...] 低 metrics.sample.window.ms 计算度量样本的时间窗口。 long 30000 [1，...] 低 quota.window.num 要在客户端配额的内存中保留的样本数 int 11 [1，...] 低 quota.window.size.seconds 客户端配额的每个样本的时间跨度 int 1 [1，...] 低 replication.quota.window.num 要在复制配额的内存中保留的样本数 int 11 [1，...] 低 replication.quota.window.size.seconds 复制配额的每个样本的时间跨度 int 1 [1，...] 低 ssl.endpoint.identification.algorithm 端点标识算法，使用服务器证书验证服务器主机名。串空值低 ssl.secure.random.implementation 用于SSL加密操作的SecureRandom PRNG实现。串空值低 zookeeper.sync.time.ms ZK领导者背后有多远ZK追随者 int 2000年低

在斯卡拉类中找到有关代理配置的详细信息

kafka.server.KafkaConfig

。

主题级配置的配置有关的话题既有服务器的默认，以及一个可选的每个话题覆盖。如果没有给出每主题配置，则使用服务器默认值。覆盖可以在主题创建时给予的一个或多个设置

--config

选项。此示例创建一个名为话题我-话题与自定义最大邮件大小和冲洗率：

 > bin / kafka-topics.sh --zookeeper localhost：2181 --create --topic my-topic --partitions 1
          --replication-factor 1 --config max.message.bytes = 64000 --config flush.messages = 1

也可以使用alter configs命令更改或设置覆盖。此示例更新最大邮件大小为我-话题：

 > bin / kafka-configs.sh --zookeeper localhost：2181 --entity-type topics --entity-name my-topic --alter --add-config max.message.bytes = 128000

要检查对主题设置的覆盖，您可以执行

 > bin / kafka-configs.sh --zookeeper localhost：2181 --entity-type topics --entity-name my-topic --describe

要删除覆盖您可以做

 > bin / kafka-configs.sh --zookeeper localhost：2181 --entity-type topics --entity-name my-topic --alter --delete-config max.message.bytes

以下是主题级别的配置。服务器的此属性的默认配置在“服务器默认属性”标题下给出。给定的服务器默认配置值仅适用于主题，如果它没有显式的主题配置覆盖。

名称

描述

类型

默认

有效值

服务器默认属性

重要性

cleanup.policy

一个字符串，“删除”或“紧凑”。此字符串指定要在旧日志段上使用的保留策略。默认策略（“删除”）会在达到保留时间或大小限制时丢弃旧细分。该“契约”的设置将使日志压实的话题。

列表

[删除]

[紧凑，删除]

log.cleanup.policy

中

compression.type

指定给定主题的最终压缩类型。此配置接受标准压缩编解码器（'gzip'，'snappy'，lz4）。它还接受“未压缩”，这相当于没有压缩; 和“生成器”，意味着保留由生产者设置的原始压缩编解码器。

串

生产者

[uncompressed，snappy，lz4，gzip，producer]

compression.type

中

delete.retention.ms

的时间量，以保留删除标记墓碑记录压实主题。此设置还给出了消费者必须完成读取的时间的界限，如果它们从偏移0开始，以确保它们获得最后阶段的有效快照（否则可以在完成扫描之前收集删除的墓碑）。

长

86400000

[0，...]

log.cleaner.delete.retention.ms

中

file.delete.delay.ms

从文件系统中删除文件之前等待的时间

长

60000

[0，...]

log.segment.delete.delay.ms

中

flush.messages

此设置允许指定我们将强制写入日志的数据的fsync的间隔。例如，如果这被设置为1，我们将在每个消息后fsync; 如果是5，我们将在每5个消息后fsync。一般来说，我们建议您不要设置此值，并使用复制持久性，并允许操作系统的后台刷新功能，因为它更高效。这个设置可以对每个话题的基础上（见覆盖的每个主题配置部分）。

长

9223372036854775807

[0，...]

log.flush.interval.messages

中

flush.ms

此设置允许指定我们将强制写入日志的数据的fsync的时间间隔。例如，如果这被设置为1000，我们将在1000毫秒过去后fsync。一般来说，我们建议您不要设置此值，并使用复制持久性，并允许操作系统的后台刷新功能，因为它更高效。

长

9223372036854775807

[0，...]

log.flush.interval.ms

中

follower.replication.throttled.replicas

在跟随者端应该限制日志复制的副本列表。该列表应该以[PartitionId]的形式描述一组副本：[BrokerId]，[PartitionId]：[BrokerId]：...或者通配符“*”可以用于限制此主题的所有副本。

列表

[]

kafka.server.ThrottledReplicaListValidator$@59d57c39

follower.replication.throttled.replicas

中

index.interval.bytes

此设置控制Kafka向其偏移索引添加索引条目的频率。默认设置确保我们大约每4096个字节对消息进行索引。更多索引允许读取更接近日志中的确切位置，但使索引更大。你可能不需要改变这个。

int

4096

[0，...]

log.index.interval.bytes

中

leader.replication.throttled.replicas

在领导者端应该限制日志复制的副本列表。该列表应该以[PartitionId]的形式描述一组副本：[BrokerId]，[PartitionId]：[BrokerId]：...或者通配符“*”可以用于限制此主题的所有副本。

列表

[]

kafka.server.ThrottledReplicaListValidator$@59d57c39

leader.replication.throttled.replicas

中

max.message.bytes

这是Kafka允许附加的最大邮件大小。请注意，如果您增加此大小，您还必须增加使用者的抓取大小，以便他们可以抓取这么大的邮件。

int

1000012

[0，...]

message.max.bytes

中

message.format.version

串

0.10.1-IV2

log.message.format.version

中

message.timestamp.difference.max.ms

代理接收消息时的时间戳和消息中指定的时间戳之间允许的最大差异。如果message.timestamp.type = CreateTime，如果时间戳的差异超过此阈值，则会拒绝邮件。如果message.timestamp.type = LogAppendTime，则忽略此配置。

长

9223372036854775807

[0，...]

log.message.timestamp.difference.max.ms

中

message.timestamp.type

定义消息中的时间戳是消息创建时间还是日志附加时间。该值应为CreateTime或LogAppendTime

串

CreateTime

log.message.timestamp.type

中

min.cleanable.dirty.ratio

此配置控制日志的压实机将如何经常尝试清理日志（假设日志压缩已启用）。默认情况下，我们将避免清除超过50％的日志已压缩的日志。此比率限制了日志中由重复项浪费的最大空间（在50％处，最多50％的日志可能是重复项）。较高的比率意味着更少，更有效的清洁，但是将意味着更多的浪费在日志中的空间。

双

0.5

[0，...，1]

log.cleaner.min.cleanable.ratio

中

min.compaction.lag.ms

消息在日志中保持未压缩的最小时间。仅适用于正在压缩的日志。

long

[0，...]

log.cleaner.min.compaction.lag.ms

中

min.insync.replicas

int

[1，...]

min.insync.replicas

中

preallocate

应该在创建新段时预分配文件？

布尔

假

log.preallocate

中

retention.bytes

此配置控制日志可以增长到的最大大小，直到我们放弃旧的日志段以释放空间，如果我们使用“删除”保留策略。默认情况下，没有大小限制只有时间限制。

long

-1

log.retention.bytes

中

retention.ms

此配置控制我们将保留日志的最大时间，直到我们放弃旧的日志段以释放空间，如果我们使用“删除”保留策略。这代表了关于消费者必须读取其数据的SLA。

long

604800000

log.retention.ms

中

segment.bytes

此配置控制日志的段文件大小。保留和清除总是一次执行一个文件，因此较大的段大小意味着更少的文件，但对保留的粒度控制较少。

int

1073741824

[14，...]

log.segment.bytes

中

segment.index.bytes

此配置控制将偏移映射到文件位置的索引的大小。我们预分配此索引文件，并收缩日志滚动后。您通常不需要更改此设置。

int

10485760

[0，...]

log.index.size.max.bytes

中

segment.jitter.ms

最大随机抖动从计划的段滚动时间减去，以避免雷击群的段滚动

long

[0，...]

log.roll.jitter.ms

中

segment.ms

此配置控制Kafka将强制日志滚动（即使分段文件未满）的时间段，以确保保留可以删除或压缩旧数据。

long

604800000

[0，...]

log.roll.ms

中

unclean.leader.election.enable

指示是否启用不在ISR集中的副本作为最后手段被选为leader，即使这样做可能会导致数据丢失

布尔

真正

unclean.leader.election.enable

中

3.2生产者配置

下面是Java生产者的配置：

名称

描述

类型

默认

有效值

重要性

bootstrap.servers

用于建立与Kafka集群的初始连接的主机/端口对的列表。客户端将使用所有服务器，而不考虑在此指定哪些服务器进行引导 - 此列表仅影响用于发现全套服务器的初始主机。这份名单应该是形式host1:port1,host2:port2,...。由于这些服务器仅用于初始连接以发现完整的集群成员资格（可能动态更改），所以此列表不需要包含全套服务器（您可能需要多个服务器，以防万一服务器关闭）。

列表

高

key.serializer

串行器类实现的关键Serializer接口。

类

高

value.serializer

串行器类实现值Serializer接口。

类

高

acks

生产者需要领导者在考虑请求完成之前已经接收的确认的数目。这控制发送的记录的持久性。允许以下设置： acks=0如果设置为零，那么生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并考虑发送。不保证可取得的服务器已收到在这种情况下，记录和retries配置不会生效（作为客户端通常不会知道任何故障）。每个记录返回的偏移将始终设置为-1。 acks=1这将意味着领导者将记录写入其本地日志，但将作出响应，而不等待所有追随者的完全确认。在这种情况下，如果领导者在确认记录之后立即失败，但在追随者复制它之前失败，则记录将丢失。 acks=all这意味着领导将等待完整的同步副本集合来确认记录。这保证只要至少一个同步中的副本保持活动，记录就不会丢失。这是最强的可用保证。这相当于acks = -1设置。

串

[all，-1,0,1]

高

buffer.memory

生产者可以用来缓冲等待发送到服务器的记录的内存的总字节数。如果记录发送速度比它们可以被传递到生产者将用于阻塞服务器max.block.ms后，它会抛出异常。此设置应大致对应于生产者将使用的总内存，但不是硬约束，因为并不是生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启用压缩）以及用于维护飞行中请求。

long

33554432

[0，...]

高

compression.type

生产者生成的所有数据的压缩类型。默认值为none（即无压缩）。有效值为none，gzip，snappy，或lz4。压缩是完整批次的数据，因此批量化的效果也将影响压缩比（更多的批次意味着更好的压缩）。

串

没有

高

retries

设置大于零的值将导致客户端重新发送任何发送失败且可能存在临时错误的记录。请注意，此重试与客户端在接收到错误时重新发送记录没有什么不同。允许重试时没有设置max.in.flight.requests.per.connection为1将潜在地改变的记录的顺序，因为如果两批被送到一个单个分区，并且第一失败并且重试，但第二成功，则在第二批中的记录可以首先出现。

int

[0，...，2147483647]

高

ssl.key.password

密钥存储文件中的私钥的密码。这对于客户端是可选的。

密码

空值

高

ssl.keystore.location

密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。

串

空值

高

ssl.keystore.password

密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。

密码

空值

高

ssl.truststore.location

信任存储文件的位置。

串

空值

高

ssl.truststore.password

信任存储文件的密码。

密码

空值

高

batch.size

每当多个记录被发送到同一分区时，生产者将尝试将批记录一起分成较少的请求。这有助于客户端和服务器上的性能。此配置控制默认批量大小（以字节为单位）。不会尝试大于此大小的批记录。发送到代理的请求将包含多个批次，每个分区一个，可用于发送的数据。小批量大小将使批处理不那么常见，并可能降低吞吐量（批量大小为零将完全禁用批处理）。非常大的批量大小可能更浪费地使用存储器，因为我们将总是分配预定额外记录的指定批量大小的缓冲器。

int

16384

[0，...]

中

client.id

在发出请求时传递给服务器的id字符串。这样做的目的是通过允许在服务器端请求记录中包括一个逻辑应用程序名称，能够跟踪超出ip / port的请求源。

串

“”

中

connections.max.idle.ms

在此配置指定的毫秒数后关闭空闲连接。

长

540000

中

linger.ms

生成器将在请求传输之间到达的任何记录集合到单个批处理请求中。通常这仅在负载下发生，当记录到达比它们可以被发送时快。然而在某些情况下，即使在中等负载下，客户端也可能希望减少请求的数量。该设置通过添加少量的人为延迟来实现这一点，即，不是立即发出记录，而是生产者将等待给定的延迟，以允许发送其他记录，使得发送可以被批处理在一起。这可以被认为类似于TCP中的Nagle算法。此设置使对配料延迟上限：一旦我们得到batch.size的一个分区记录值得将立即无论此设置发送的，但是如果我们比积累让这个分区，我们将'苟延残喘'这个字节数少指定的时间等待更多的记录显示。此设置默认为0（即无延迟）。设定linger.ms=5，例如，将具有减少发送的请求的数量的效果，但会高达延迟5ms的增加在负载由于缺少发送记录。

长

[0，...]

中

max.block.ms

配置控制多久KafkaProducer.send()和KafkaProducer.partitionsFor()将block.These方法可以阻止或者因为缓冲器是在用户提供的串行器或分割器充满或元数据unavailable.Blocking将不会针对此超时计数。

长

60000

[0，...]

中

max.request.size

请求的最大大小（以字节为单位）。这也是有效的最大记录大小的上限。请注意，服务器有自己的记录大小上限，可能与此不同。此设置将限制生产者在单个请求中发送的记录批次数，以避免发送大量请求。

int

1048576

[0，...]

中

partitioner.class

分区类实现Partitioner接口。

类

class org.apache.kafka.clients.producer.internals.DefaultPartitioner

中

receive.buffer.bytes

读取数据时使用的TCP接收缓冲区大小（SO_RCVBUF）。如果值为-1，将使用操作系统默认值。

int

32768

[-1，...]

中

request.timeout.ms

配置控制客户端等待请求响应的最大时间量。如果在超时时间之前没有收到响应，客户端将在必要时重新发送请求，如果重试次数耗尽，则请求失败。

int

30000

[0，...]

中

sasl.kerberos.service.name

Kafka运行的Kerberos主体名称。这可以在Kafka的JAAS配置或Kafka的配置中定义。

串

空值

中

sasl.mechanism

SASL机制用于客户端连接。这可以是安全提供者可用的任何机制。GSSAPI是默认机制。

串

GSSAPI

中

security.protocol

用于与代理通信的协议。有效值为：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL。

串

纯文本

中

send.buffer.bytes

发送数据时使用的TCP发送缓冲区（SO_SNDBUF）的大小。如果值为-1，将使用操作系统默认值。

int

131072

[-1，...]

中

ssl.enabled.protocols

为SSL连接启用的协议列表。

列表

[TLSv1.2，TLSv1.1，TLSv1]

中

ssl.keystore.type

密钥存储文件的文件格式。这对于客户端是可选的。

串

JKS

中

ssl.protocol

串

TLS

中

ssl.provider

用于SSL连接的安全提供程序的名称。默认值是JVM的默认安全提供程序。

串

空值

中

ssl.truststore.type

信任存储文件的文件格式。

串

JKS

中

timeout.ms

配置控制的时候，服务器会等待确认从追随者到满足确认的要求生产者已与规定的最高限额acks配置。如果在超时过期时未满足所请求的确认数量，则将返回错误。此超时在服务器端测量，不包括请求的网络延迟。

int

30000

[0，...]

中

block.on.buffer.full

当我们的内存缓冲区用尽时，我们必须停止接受新的记录（块）或抛出错误。默认情况下此设置为false，并且生产者将不再抛出BufferExhaustException而是会用max.block.ms价值来阻止，之后它会抛出一个TimeoutException。此属性设置为true将设置max.block.ms为Long.MAX_VALUE。此外，如果该属性设置为true，参数metadata.fetch.timeout.ms不再荣幸。此参数已弃用，将在以后的版本中删除。参数max.block.ms应使用。

布尔

假

低

interceptor.classes

要用作拦截器的类的列表。实现ProducerInterceptor界面，您可以拦截（也可能发生变异）由生产者收到的记录，他们都发布到卡夫卡群集之前。默认情况下，没有拦截器。

列表

空值

低

max.in.flight.requests.per.connection

客户端在阻止之前在单个连接上发送的未确认请求的最大数量。请注意，如果此设置设置为大于1并且发送失败，则可能会由于重试（即启用重试）而导致消息重新排序。

int

[1，...]

低

metadata.fetch.timeout.ms

第一次将数据发送到主题时，我们必须获取有关该主题的元数据，以了解哪些服务器托管主题的分区。此配置指定在将异常返回到客户端之前此次提取成功的最大时间（以毫秒为单位）。

长

60000

[0，...]

低

metadata.max.age.ms

即使我们没有看到任何分区领导更改以主动发现任何新的代理或分区，我们强制刷新元数据的时间（以毫秒为单位）。

长

300000

[0，...]

低

metric.reporters

用作度量报告器的类的列表。实现MetricReporter接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。

列表

[]

低

metrics.num.samples

维持计算度量的样本数。

int

[1，...]

低

metrics.sample.window.ms

计算度量样本的时间窗口。

长

30000

[0，...]

低

reconnect.backoff.ms

尝试重新连接到给定主机之前等待的时间。这避免了在紧密环路中重复连接到主机。此回退适用于消费者发送给代理的所有请求。

长

[0，...]

低

retry.backoff.ms

尝试对指定主题分区重试失败的请求之前等待的时间。这避免了在一些故障情况下在紧密循环中重复发送请求。

长

100

[0，...]

低

sasl.kerberos.kinit.cmd

Kerberos kinit命令路径。

串

/ usr / bin / kinit

低

sasl.kerberos.min.time.before.relogin

登录线程在刷新尝试之间的休眠时间。

长

60000

低

sasl.kerberos.ticket.renew.jitter

添加到更新时间的随机抖动的百分比。

双

0.05

低

sasl.kerberos.ticket.renew.window.factor

登录线程将睡眠，直到从上次刷新到票据到期的时间的指定窗口因子已经到达，在该时间它将尝试更新票证。

双

0.8

低

ssl.ciphersuites

列表

空值

低

ssl.endpoint.identification.algorithm

端点标识算法，使用服务器证书验证服务器主机名。

串

空值

低

ssl.keymanager.algorithm

密钥管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。

串

SunX509

低

ssl.secure.random.implementation

用于SSL加密操作的SecureRandom PRNG实现。

串

空值

低

ssl.trustmanager.algorithm

信任管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的信任管理器工厂算法。

串

PKIX

低

acks=0如果设置为零，那么生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并考虑发送。不保证可取得的服务器已收到在这种情况下，记录和retries配置不会生效（作为客户端通常不会知道任何故障）。每个记录返回的偏移将始终设置为-1。
acks=1这将意味着领导者将记录写入其本地日志，但将作出响应，而不等待所有追随者的完全确认。在这种情况下，如果领导者在确认记录之后立即失败，但在追随者复制它之前失败，则记录将丢失。
acks=all这意味着领导将等待完整的同步副本集合来确认记录。这保证只要至少一个同步中的副本保持活动，记录就不会丢失。这是最强的可用保证。这相当于acks = -1设置。

串 1 [all，-1,0,1] 高 buffer.memory 生产者可以用来缓冲等待发送到服务器的记录的内存的总字节数。如果记录发送速度比它们可以被传递到生产者将用于阻塞服务器

max.block.ms

后，它会抛出异常。此设置应大致对应于生产者将使用的总内存，但不是硬约束，因为并不是生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启用压缩）以及用于维护飞行中请求。 long 33554432 [0，...] 高 compression.type 生产者生成的所有数据的压缩类型。默认值为none（即无压缩）。有效值为

none

，

gzip

，

snappy

，或

lz4

。压缩是完整批次的数据，因此批量化的效果也将影响压缩比（更多的批次意味着更好的压缩）。串没有高 retries 设置大于零的值将导致客户端重新发送任何发送失败且可能存在临时错误的记录。请注意，此重试与客户端在接收到错误时重新发送记录没有什么不同。允许重试时没有设置

max.in.flight.requests.per.connection

为1将潜在地改变的记录的顺序，因为如果两批被送到一个单个分区，并且第一失败并且重试，但第二成功，则在第二批中的记录可以首先出现。 int 0 [0，...，2147483647] 高 ssl.key.password 密钥存储文件中的私钥的密码。这对于客户端是可选的。密码空值高 ssl.keystore.location 密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。串空值高 ssl.keystore.password 密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。密码空值高 ssl.truststore.location 信任存储文件的位置。串空值高 ssl.truststore.password 信任存储文件的密码。密码空值高 batch.size 每当多个记录被发送到同一分区时，生产者将尝试将批记录一起分成较少的请求。这有助于客户端和服务器上的性能。此配置控制默认批量大小（以字节为单位）。不会尝试大于此大小的批记录。发送到代理的请求将包含多个批次，每个分区一个，可用于发送的数据。小批量大小将使批处理不那么常见，并可能降低吞吐量（批量大小为零将完全禁用批处理）。非常大的批量大小可能更浪费地使用存储器，因为我们将总是分配预定额外记录的指定批量大小的缓冲器。 int 16384 [0，...] 中 client.id 在发出请求时传递给服务器的id字符串。这样做的目的是通过允许在服务器端请求记录中包括一个逻辑应用程序名称，能够跟踪超出ip / port的请求源。串 “” 中 connections.max.idle.ms 在此配置指定的毫秒数后关闭空闲连接。长 540000 中 linger.ms 生成器将在请求传输之间到达的任何记录集合到单个批处理请求中。通常这仅在负载下发生，当记录到达比它们可以被发送时快。然而在某些情况下，即使在中等负载下，客户端也可能希望减少请求的数量。该设置通过添加少量的人为延迟来实现这一点，即，不是立即发出记录，而是生产者将等待给定的延迟，以允许发送其他记录，使得发送可以被批处理在一起。这可以被认为类似于TCP中的Nagle算法。此设置使对配料延迟上限：一旦我们得到

batch.size

的一个分区记录值得将立即无论此设置发送的，但是如果我们比积累让这个分区，我们将'苟延残喘'这个字节数少指定的时间等待更多的记录显示。此设置默认为0（即无延迟）。设定

linger.ms=5

，例如，将具有减少发送的请求的数量的效果，但会高达延迟5ms的增加在负载由于缺少发送记录。长 0 [0，...] 中 max.block.ms 配置控制多久

KafkaProducer.send()

和

KafkaProducer.partitionsFor()

将block.These方法可以阻止或者因为缓冲器是在用户提供的串行器或分割器充满或元数据unavailable.Blocking将不会针对此超时计数。长 60000 [0，...] 中 max.request.size 请求的最大大小（以字节为单位）。这也是有效的最大记录大小的上限。请注意，服务器有自己的记录大小上限，可能与此不同。此设置将限制生产者在单个请求中发送的记录批次数，以避免发送大量请求。 int 1048576 [0，...] 中 partitioner.class 分区类实现

Partitioner

接口。类 class org.apache.kafka.clients.producer.internals.DefaultPartitioner 中 receive.buffer.bytes 读取数据时使用的TCP接收缓冲区大小（SO_RCVBUF）。如果值为-1，将使用操作系统默认值。 int 32768 [-1，...] 中 request.timeout.ms 配置控制客户端等待请求响应的最大时间量。如果在超时时间之前没有收到响应，客户端将在必要时重新发送请求，如果重试次数耗尽，则请求失败。 int 30000 [0，...] 中 sasl.kerberos.service.name Kafka运行的Kerberos主体名称。这可以在Kafka的JAAS配置或Kafka的配置中定义。串空值中 sasl.mechanism SASL机制用于客户端连接。这可以是安全提供者可用的任何机制。GSSAPI是默认机制。串 GSSAPI 中 security.protocol 用于与代理通信的协议。有效值为：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL。串纯文本中 send.buffer.bytes 发送数据时使用的TCP发送缓冲区（SO_SNDBUF）的大小。如果值为-1，将使用操作系统默认值。 int 131072 [-1，...] 中 ssl.enabled.protocols 为SSL连接启用的协议列表。列表 [TLSv1.2，TLSv1.1，TLSv1] 中 ssl.keystore.type 密钥存储文件的文件格式。这对于客户端是可选的。串 JKS 中 ssl.protocol 用于生成SSLContext的SSL协议。默认设置为TLS，这在大多数情况下是正常的。最近的JVM中允许的值为TLS，TLSv1.1和TLSv1.2。在较早的JVM中可能支持SSL，SSLv2和SSLv3，但是由于已知的安全漏洞，它们的使用不受欢迎。串 TLS 中 ssl.provider 用于SSL连接的安全提供程序的名称。默认值是JVM的默认安全提供程序。串空值中 ssl.truststore.type 信任存储文件的文件格式。串 JKS 中 timeout.ms 配置控制的时候，服务器会等待确认从追随者到满足确认的要求生产者已与规定的最高限额

acks

配置。如果在超时过期时未满足所请求的确认数量，则将返回错误。此超时在服务器端测量，不包括请求的网络延迟。 int 30000 [0，...] 中 block.on.buffer.full 当我们的内存缓冲区用尽时，我们必须停止接受新的记录（块）或抛出错误。默认情况下此设置为false，并且生产者将不再抛出BufferExhaustException而是会用

max.block.ms

价值来阻止，之后它会抛出一个TimeoutException。此属性设置为true将设置

max.block.ms

为Long.MAX_VALUE。此外，如果该属性设置为true，参数

*metadata.fetch.timeout.ms*

不再荣幸。 此参数已弃用，将在以后的版本中删除。参数

max.block.ms

应使用。布尔假低 interceptor.classes 要用作拦截器的类的列表。实现

ProducerInterceptor

界面，您可以拦截（也可能发生变异）由生产者收到的记录，他们都发布到卡夫卡群集之前。默认情况下，没有拦截器。列表空值低 max.in.flight.requests.per.connection 客户端在阻止之前在单个连接上发送的未确认请求的最大数量。请注意，如果此设置设置为大于1并且发送失败，则可能会由于重试（即启用重试）而导致消息重新排序。 int 5 [1，...] 低 metadata.fetch.timeout.ms 第一次将数据发送到主题时，我们必须获取有关该主题的元数据，以了解哪些服务器托管主题的分区。此配置指定在将异常返回到客户端之前此次提取成功的最大时间（以毫秒为单位）。长 60000 [0，...] 低 metadata.max.age.ms 即使我们没有看到任何分区领导更改以主动发现任何新的代理或分区，我们强制刷新元数据的时间（以毫秒为单位）。长 300000 [0，...] 低 metric.reporters 用作度量报告器的类的列表。实现

MetricReporter

接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。列表 [] 低 metrics.num.samples 维持计算度量的样本数。 int 2 [1，...] 低 metrics.sample.window.ms 计算度量样本的时间窗口。长 30000 [0，...] 低 reconnect.backoff.ms 尝试重新连接到给定主机之前等待的时间。这避免了在紧密环路中重复连接到主机。此回退适用于消费者发送给代理的所有请求。长 50 [0，...] 低 retry.backoff.ms 尝试对指定主题分区重试失败的请求之前等待的时间。这避免了在一些故障情况下在紧密循环中重复发送请求。长 100 [0，...] 低 sasl.kerberos.kinit.cmd Kerberos kinit命令路径。串 / usr / bin / kinit 低 sasl.kerberos.min.time.before.relogin 登录线程在刷新尝试之间的休眠时间。长 60000 低 sasl.kerberos.ticket.renew.jitter 添加到更新时间的随机抖动的百分比。双 0.05 低 sasl.kerberos.ticket.renew.window.factor 登录线程将睡眠，直到从上次刷新到票据到期的时间的指定窗口因子已经到达，在该时间它将尝试更新票证。双 0.8 低 ssl.ciphersuites 密码套件列表。这是一种命名的认证，加密，MAC和密钥交换算法的组合，用于使用TLS或SSL网络协议协商网络连接的安全设置。默认情况下，支持所有可用的密码套件。列表空值低 ssl.endpoint.identification.algorithm 端点标识算法，使用服务器证书验证服务器主机名。串空值低 ssl.keymanager.algorithm 密钥管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。串 SunX509 低 ssl.secure.random.implementation 用于SSL加密操作的SecureRandom PRNG实现。串空值低 ssl.trustmanager.algorithm 信任管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的信任管理器工厂算法。串 PKIX 低

对于那些有兴趣在旧的scala生产者CONFIGS，信息可以找到从这里。

3.3消费者配置

在0.9.0.0中，我们引入了新的Java消费者作为替代旧的基于Scala的简单和高级消费者。新用户和旧用户的配置如下所述。

3.3.1新的消费者配置

以下是新客户的配置：

名称

描述

类型

默认

有效值

重要性

bootstrap.servers

列表

高

key.deserializer

解串器类实现的关键Deserializer接口。

类

高

value.deserializer

解串器类实现值Deserializer接口。

类

高

fetch.min.bytes

服务器应该为获取请求返回的最小数据量。如果数据不足，请求将在应答请求之前等待多少数据累积。默认设置为1字节表示只要单个字节的数据可用或获取请求超时等待数据到达，就会应答获取请求。将此值设置为大于1的值将导致服务器等待大量数据累积，这可能以一些额外延迟为代价提高服务器吞吐量。

int

[0，...]

高

group.id

标识此消费者所属的使用者组的唯一字符串。如果消费者通过使用该组的管理功能，则需要此属性subscribe(topic)或基于卡夫卡偏移管理策略。

串

“”

高

heartbeat.interval.ms

使用Kafka的组管理设施时，心跳到消费者协调器之间的预期时间。心跳用于确保消费者的会话保持活动并且当新消费者加入或离开组时促进重新平衡。值必须设定为比低session.timeout.ms，但通常应设置不高于该值的1/3。它可以调整得更低，以控制正常再平衡的预期时间。

int

3000

高

max.partition.fetch.bytes

服务器将返回的每个分区的最大数据量。如果提取的第一个非空分区中的第一条消息大于此限制，则仍会返回消息以确保消费者可以取得进展。由broker接受的最大邮件大小是通过定义message.max.bytes（broker配置）或max.message.bytes（主题配置）。请参阅fetch.max.bytes以限制使用者请求大小

int

1048576

[0，...]

高

session.timeout.ms

用于在使用Kafka的组管理工具时检测用户故障的超时。消费者发送周期性心跳以向代理指示其活跃度。如果在此会话超时期满之前代理没有收到心跳，则代理将从组中删除此使用者并启动重新平衡。注意，该值必须是在允许范围内所配置的代理配置由group.min.session.timeout.ms和group.max.session.timeout.ms。

int

10000

高

ssl.key.password

密钥存储文件中的私钥的密码。这对于客户端是可选的。

密码

空值

高

ssl.keystore.location

密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。

串

空值

高

ssl.keystore.password

密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。

密码

空值

高

ssl.truststore.location

信任存储文件的位置。

串

空值

高

ssl.truststore.password

信任存储文件的密码。

密码

空值

高

auto.offset.reset

当Kafka中没有初始偏移或如果当前??偏移在服务器上不再存在时（例如，因为该数据已被删除），该怎么办：最早：自动将偏移重置为最早的偏移最新：自动将偏移重置为最新偏移 none：如果没有为消费者组找到以前的偏移，则向消费者抛出异常任何其他：抛出异常到消费者。

串

3.4 Kafka连接配置

下面是Kafka Connect框架的配置。

名称

描述

类型

默认

有效值

重要性

config.storage.topic

kafka主题来存储配置

串

高

group.id

标识此工人所属的Connect群集组的唯一字符串。

串

高

key.converter

Converter类用于在Kafka Connect格式和写入Kafka的序列化格式之间进行转换。这控制写入或读取Kafka的消息中的键的格式，并且由于它独立于连接器，因此它允许任何连接器使用任何序列化格式。常见格式的示例包括JSON和Avro。

类

高

offset.storage.topic

kafka主题来存储连接器偏移量

串

高

status.storage.topic

kafka主题来跟踪连接器和任务状态

串

高

value.converter

Converter类用于在Kafka Connect格式和写入Kafka的序列化格式之间进行转换。这控制写入或读取Kafka的消息中的值的格式，并且由于它独立于连接器，因此它允许任何连接器使用任何序列化格式。常见格式的示例包括JSON和Avro。

类

高

internal.key.converter

Converter类用于在Kafka Connect格式和写入Kafka的序列化格式之间进行转换。这控制写入或读取Kafka的消息中的键的格式，并且由于它独立于连接器，因此它允许任何连接器使用任何序列化格式。常见格式的示例包括JSON和Avro。此设置控制用于框架使用的内部记录数据的格式（如配置和偏移量），因此用户通常可以使用任何正常运行的Converter实现。

类

低

internal.value.converter

Converter类用于在Kafka Connect格式和写入Kafka的序列化格式之间进行转换。这控制写入或读取Kafka的消息中的值的格式，并且由于它独立于连接器，因此它允许任何连接器使用任何序列化格式。常见格式的示例包括JSON和Avro。此设置控制用于框架使用的内部记录数据的格式（如配置和偏移量），因此用户通常可以使用任何正常运行的Converter实现。

类

低

bootstrap.servers

列表

[localhost：9092]

高

heartbeat.interval.ms

使用Kafka的组管理设施时，心跳到组协调器之间的预计时间。心跳用于确保工作者的会话保持活动状态，并在新成员加入或离开组时方便重新平衡。值必须设定为比低session.timeout.ms，但通常应设置不高于该值的1/3。它可以调整得更低，以控制正常再平衡的预期时间。

int

3000

高

rebalance.timeout.ms

重新平衡开始后每个工作人员加入群组的最长允许时间。这基本上是对所有任务刷新任何挂起的数据和提交偏移量所需的时间量的限制。如果超过超时，那么工作程序将从组中删除，这将导致偏移提交失败。

int

60000

高

session.timeout.ms

用于检测worker失败的超时。工作者发送定期心跳以向代理指示其活跃度。如果代理在此会话超时期满之前未收到心跳，则代理将从组中删除工作程序并启动重新平衡。注意，该值必须是在允许范围内所配置的代理配置由group.min.session.timeout.ms和group.max.session.timeout.ms。

int

10000

高

ssl.key.password

密钥存储文件中的私钥的密码。这对于客户端是可选的。

密码

空值

高

ssl.keystore.location

密钥存储文件的位置。这对于客户端是可选的，并且可以用于客户端的双向认证。

串

空值

高

ssl.keystore.password

密钥存储文件的存储密码。这对于客户端是可选的，只有在配置了ssl.keystore.location时才需要。

密码

空值

高

ssl.truststore.location

信任存储文件的位置。

串

空值

高

ssl.truststore.password

信任存储文件的密码。

密码

空值

高

connections.max.idle.ms

在此配置指定的毫秒数后关闭空闲连接。

长

540000

中

receive.buffer.bytes

读取数据时使用的TCP接收缓冲区大小（SO_RCVBUF）。如果值为-1，将使用操作系统默认值。

int

32768

[0，...]

中

request.timeout.ms

配置控制客户端等待请求响应的最大时间量。如果在超时时间之前没有收到响应，客户端将在必要时重新发送请求，如果重试次数耗尽，则请求失败。

int

40000

[0，...]

中

sasl.kerberos.service.name

Kafka运行的Kerberos主体名称。这可以在Kafka的JAAS配置或Kafka的配置中定义。

串

空值

中

sasl.mechanism

SASL机制用于客户端连接。这可以是安全提供者可用的任何机制。GSSAPI是默认机制。

串

GSSAPI

中

security.protocol

用于与代理通信的协议。有效值为：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL。

串

纯文本

中

send.buffer.bytes

发送数据时使用的TCP发送缓冲区（SO_SNDBUF）的大小。如果值为-1，将使用操作系统默认值。

int

131072

[0，...]

中

ssl.enabled.protocols

为SSL连接启用的协议列表。

列表

[TLSv1.2，TLSv1.1，TLSv1]

中

ssl.keystore.type

密钥存储文件的文件格式。这对于客户端是可选的。

串

JKS

中

ssl.protocol

串

TLS

中

ssl.provider

用于SSL连接的安全提供程序的名称。默认值是JVM的默认安全提供程序。

串

空值

中

ssl.truststore.type

信任存储文件的文件格式。

串

JKS

中

worker.sync.timeout.ms

当工作程序与其他工作程序不同步并且需要重新同步配置时，请等待到这个时间量，然后放弃，离开组，并在重新加入之前等待一个后退周期。

int

3000

中

worker.unsync.backoff.ms

当工作程序与其他工作程序不同步并且未能在worker.sync.timeout.ms中赶上时，在重新加入之前，将Connect集群保留此时间很长时间。

int

300000

中

access.control.allow.methods

通过设置Access-Control-Allow-Methods头设置跨源请求支持的方法。Access-Control-Allow-Methods标头的默认值允许GET，POST和HEAD的跨源请求。

串

“”

低

access.control.allow.origin

为REST API请求设置Access-Control-Allow-Origin标头的值。要启用跨源访问，请将此设置为应允许访问API的应用程序的域，或'*'以允许从任何域。默认值仅允许从REST API的域进行访问。

串

“”

低

client.id

在发出请求时传递给服务器的id字符串。这样做的目的是通过允许在服务器端请求记录中包括一个逻辑应用程序名称，能够跟踪超出ip / port的请求源。

串

“”

低

metadata.max.age.ms

即使我们没有看到任何分区领导更改以主动发现任何新的代理或分区，我们强制刷新元数据的时间（以毫秒为单位）。

长

300000

[0，...]

低

metric.reporters

用作度量报告器的类的列表。实现MetricReporter接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。

列表

[]

低

metrics.num.samples

维持计算度量的样本数。

int

[1，...]

低

metrics.sample.window.ms

计算度量样本的时间窗口。

长

30000

[0，...]

低

offset.flush.interval.ms

尝试提交任务的偏移量的间隔。

长

60000

低

offset.flush.timeout.ms

在取消过程之前等待记录刷新并将偏移数据分区为偏移存储的最大毫秒数，并恢复在未来尝试中提交的偏移数据。

长

5000

低

reconnect.backoff.ms

尝试重新连接到给定主机之前等待的时间。这避免了在紧密环路中重复连接到主机。此回退适用于消费者发送给代理的所有请求。

长

[0，...]

低

rest.advertised.host.name

如果设置，这是将给予其他工作人员连接到的主机名。

串

空值

低

rest.advertised.port

如果设置，这是将给予其他工作人员连接到的端口。

int

空值

低

rest.host.name

REST API的主机名。如果设置，它将只绑定到此接口。

串

空值

低

rest.port

用于REST API的端口。

int

8083

低

retry.backoff.ms

尝试对指定主题分区重试失败的请求之前等待的时间。这避免了在一些故障情况下在紧密循环中重复发送请求。

长

100

[0，...]

低

sasl.kerberos.kinit.cmd

Kerberos kinit命令路径。

串

/ usr / bin / kinit

低

sasl.kerberos.min.time.before.relogin

登录线程在刷新尝试之间的休眠时间。

长

60000

低

sasl.kerberos.ticket.renew.jitter

添加到更新时间的随机抖动的百分比。

双

0.05

低

sasl.kerberos.ticket.renew.window.factor

登录线程将睡眠，直到从上次刷新到票据到期的时间的指定窗口因子已经到达，在该时间它将尝试更新票证。

双

0.8

低

ssl.ciphersuites

列表

空值

低

ssl.endpoint.identification.algorithm

端点标识算法，使用服务器证书验证服务器主机名。

串

空值

低

ssl.keymanager.algorithm

密钥管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。

串

SunX509

低

ssl.secure.random.implementation

用于SSL加密操作的SecureRandom PRNG实现。

串

空值

低

ssl.trustmanager.algorithm

信任管理器工厂用于SSL连接的算法。默认值是为Java虚拟机配置的信任管理器工厂算法。

串

PKIX

低

task.shutdown.graceful.timeout.ms

等待任务正常关闭的时间。这是总时间量，而不是每个任务。所有任务已关闭触发，然后他们按顺序等待。

长

5000

低

3.5 Kafka Streams配置

下面是Kafka Streams客户端库的配置。

名称

描述

类型

默认

有效值

重要性

application.id

流处理应用程序的标识符。在Kafka集群中必须是唯一的。它用作1）默认的client-id前缀，2）用于成员资格管理的group-id，3）changelog主题前缀。

串

高

bootstrap.servers

列表

高

client.id

在发出请求时传递给服务器的id字符串。这样做的目的是通过允许在服务器端请求记录中包括一个逻辑应用程序名称，能够跟踪超出ip / port的请求源。

串

“”

高

zookeeper.connect

Zookeeper连接字符串用于Kafka主题管理。

串

“”

高

key.serde

串行器/解串器类实现键Serde接口。

类

class org.apache.kafka.common.serialization.Serdes $ ByteArraySerde

中

partition.grouper

分区石斑鱼类，它实现的PartitionGrouper接口。

类

class org.apache.kafka.streams.processor.DefaultPartitionGrouper

中

replication.factor

由流处理应用程序创建的更改日志主题和重新分区主题的复制因素。

int

中

state.dir

状态存储的目录位置。

串

/ tmp / kafka-streams

中

timestamp.extractor

时间戳提取类，它实现的TimestampExtractor接口。

类

class org.apache.kafka.streams.processor.ConsumerRecordTimestampExtractor

中

value.serde

串行器/解串器类实现值Serde接口。

类

class org.apache.kafka.common.serialization.Serdes $ ByteArraySerde

中

windowstore.changelog.additional.retention.ms

添加到windows的maintainMs，以确保数据不会被过早地从日志中删除。允许时钟漂移。默认值为1天

long

86400000

中

application.server

指向嵌入式用户定义端点的主机：端口对，可用于在单个KafkaStreams应用程序中发现状态存储的位置

串

“”

低

buffered.records.per.partition

每个分区缓冲的最大记录数。

int

1000

低

cache.max.bytes.buffering

要用于所有线程缓冲的最大内存字节数

long

10485760

[0，...]

低

commit.interval.ms

保存处理器位置的频率。

long

30000

低

metric.reporters

用作度量报告器的类的列表。实现MetricReporter接口允许在将通知新的度量创作类堵塞。总是包括JmxReporter以注册JMX统计信息。

列表

[]

低

metrics.num.samples

维持计算度量的样本数。

int

[1，...]

低

metrics.sample.window.ms

计算度量样本的时间窗口。

long

30000

[0，...]

低

num.standby.replicas

每个任务的备用副本数。

int

低

num.stream.threads

执行流处理的线程数。

int

低

poll.ms

阻止等待输入的时间（以毫秒为单位）。

long

100

低

rocksdb.config.setter

实现一个岩石DB配置二传手类RocksDBConfigSetter接口

类

空值

低

state.cleanup.delay.ms

迁移分区时删除状态之前等待的时间（以毫秒为单位）。

long

60000

低

4.设计

4.1动机

我们设计卡夫卡是能够作为处理所有实时数据馈送的统一平台的大公司可能有。为了做到这一点，我们必须考虑一个相当广泛的用例集。

它必须具有高吞吐量以支持高容量事件流，例如实时日志聚合。

它需要优雅地处理大型数据积压，以便能够支持来自离线系统的定期数据加载。

这也意味着系统将必须处理低延迟传递以处理更传统的消息使用情况。

我们希望支持对这些订阅源的分区，分布式实时处理，以创建新的派生订阅源。这激发了我们的分区和消费模式。

最后，在流被馈送到其他数据系统用于服务的情况下，我们知道该系统将必须能够在存在机器故障的情况下保证容错。

支持这些用途导致我们设计了一些独特的元素，类似于数据库日志比传统的消息传递系统。我们将在以下部分中概述设计的一些要素。

4.2持久性

不要害怕文件系统！

Kafka很大程度上依赖于用于存储和缓存消息的文件系统。有一个普遍的看法，“磁盘是慢”，这使得人们怀疑，持久的结构可以提供竞争性能。事实上，磁盘的速度比人们期望的要慢得多，这取决于它们的使用方式; 并且正确设计的盘结构通常可以与网络一样快。

关于磁盘性能的关键事实是硬盘驱动器的吞吐量已经偏离了过去十年的磁盘寻道的延迟。其结果是线性的性能上的写入JBOD 六7200rpm的SATA RAID-5阵列配置为约600MB / sec，但随机写入的性能仅约100K /秒-一个超过6000X差异。这些线性读取和写入是所有使用模式中最可预测的，并且由操作系统进行了大量优化。现代操作系统提供预读和后写技术，以大块多次预取数据并将较小的逻辑写入分组为大型物理写入。这个问题的进一步讨论可以在这个中找到ACM队列的文章 ; 他们实际上发现连续的磁盘访问可以在某些情况下会比随机内存访问速度更快！

为了补偿这种性能分歧，现代操作系统在使用用于盘缓存的主存储器方面变得越来越积极。一个现代化的操作系统将愉快地转移所有可用内存来缓存磁盘少的性能损失，当内存被回收。所有磁盘读取和写入都将通过此统一高速缓存。如果不使用直接I / O，这个功能不能轻易关闭，因此即使一个进程维护了数据的进程内缓存，这些数据也可能在操作系统页面缓存中重复，有效地存储所有内容两次。

此外，我们正在构建在JVM之上，任何花费任何时间使用Java内存使用的人都知道两件事：

对象的内存开销非常高，通常使存储的数据大小加倍（或更差）。
Java垃圾收集随着堆内数据增加而变得越来越紧张和缓慢。

由于使用文件系统和依靠pagecache的这些因素优于维护内存中缓存或其他结构 - 我们通过自动访问所有可用内存，至少使可用缓存增加一倍，并且通过存储一个紧凑型字节结构而不是单个对象。这样做将导致高达28-30GB的缓存在32GB机器没有GC惩罚。此外，即使服务被重新启动，这个高速缓存将保持温暖，而在进程高速缓存将需要在内存中重建（对于10GB高速缓存可能需要10分钟），否则它将需要从完全冷高速缓存开始（这可能意味着可怕的初始性能）。这也大大简化了代码，因为用于保持高速缓存和文件系统之间的一致性的所有逻辑现在都在OS中，这倾向于比一次性进程中尝试更有效和更正确地进行。如果您的磁盘使用偏好线性读取，则预读在每个磁盘读取时有效地预先填充该缓存有用的数据。

这意味着一个非常简单的设计：而不是在内存中保持尽可能多的内存，并且当我们用尽空间时将它全部刷新到文件系统中，我们反过来。所有数据立即写入文件系统上的持久性日志，而不必刷新到磁盘。实际上，这意味着它被转移到内核的页面缓存中。

页缓存为中心的设计的这种风格的描述文章（以及傲慢的健康剂量）上光油这里的设计。

恒定时间

消息系统中使用的持久性数据结构通常是具有相关联的BTree或其他通用随机访问数据结构的每消费者队列，以维护关于消息的元数据。BTrees是可用的最通用的数据结构，并且使得可以在消息系统中支持各种各样的事务和非事务语义。他们确实带来了相当高的成本，但是：Btree操作是O（log N）。通常O（log N）被认为基本上等于恒定时间，但是对于磁盘操作这不是真的。磁盘寻找在10毫秒弹出，每个磁盘每次只能做一次寻找，因此并行性有限。因此，即使少数磁盘寻求导致非常高的开销。由于存储系统将非常快速的缓存操作与非常慢的物理磁盘操作混合在一起，所以树结构的观察性能通常是超级线性的，因为数据随着固定缓存增加 - 即将数据加倍使事情比两倍慢。

直观地，持久队列可以建立在简单的读取和追加到文件，通常是日志解决方案的情况。该结构具有所有操作都是O（1）和读取不阻塞写入或彼此的优点。这具有明显的性能优势，因为性能与数据大小完全解耦 - 一个服务器现在可以充分利用多个便宜的低转速1 + TB SATA驱动器。尽管它们的寻道性能较差，但是这些驱动器对于大型读取和写入具有可接受的性能，并且以1/3的价格和3倍的容量提供。

访问几乎无限的磁盘空间而没有任何性能损失意味着我们可以提供一些通常在消息传递系统中不常见的功能。例如，在Kafka中，我们可以在消息被消费之后立即删除消息，而不是尝试删除消息，而是可以在较长时间（例如一周）内保留消息。这为消费者带来了很大的灵活性，我们将在下面进行描述。

4.3效率

我们已经付出了巨大的努力效率。我们的主要用例之一是处理网络活动数据，这是非常高的量：每个页面视图可能会生成几十个写入。此外，我们假设每个发布的消息由至少一个消费者（通常很多）阅读，因此我们努力使消费尽可能廉价。

我们还发现，从构建和运行多个类似系统的经验来看，效率是有效的多租户操作的关键。如果下游基础设施服务可能由于应用程序的使用中的小的突变而容易地成为瓶颈，则这种小的改变通常会产生问题。通过非常快，我们帮助确保应用程序在基础设施之前在负载下翻转。当尝试运行在集中式集群上支持数十或数百个应用程序的集中式服务时，这是特别重要的，因为使用模式的变化是近乎每天发生的。

我们在上一节讨论磁盘效率。一旦不良磁盘访问模式被消除，在这种类型的系统中有两种常见的低效率原因：太多的小I / O操作和过多的字节复制。

小的I / O问题发生在客户端和服务器之间以及服务器自己的持久操作中。

为了避免这种情况，我们的协议是围绕一个“消息集”抽象自然地将消息组合在一起。这允许网络请求将消息分组在一起，并且摊销网络往返的开销，而不是一次发送单个消息。服务器又将消息块一次性附加到其日志，消费者一次获取大的线性块。

这种简单的优化产生数量级的加速。批处理导致更大的网络包，更大的顺序磁盘操作，连续内存块等，所有这些都允许Kafka将随机消息写入的突发流转换为流向消费者的线性写入。

另一个低效率是在字节复制。在低消息速率这不是一个问题，但在负载下的影响是显着的。为了避免这种情况，我们采用由生产者，代理和消费者共享的标准化二进制消息格式（因此，数据块可以在它们之间无修改地传输）。

代理维护的消息日志本身只是一个文件的目录，每个文件由一系列已经以生产者和消费者使用的相同格式写入磁盘的消息集填充。维护这种通用格式允许优化最重要的操作：持久日志块的网络传输。现代unix操作系统提供高度优化的代码路径，用于将数据从页面缓存传输到套接字; 在Linux中这是与完成的sendfile系统调用。

要理解sendfile的影响，理解数据从文件传输到套接字的公共数据路径很重要：

操作系统将数据从磁盘读入内核空间中的页缓存
应用程序将数据从内核空间读入用户空间缓冲区
应用程序将数据写回到内核空间到套接字缓冲区中
操作系统将数据从套接字缓冲区复制到NIC缓冲区，并通过网络发送

这显然是低效的，有四个副本和两个系统调用。使用sendfile，通过允许操作系统将数据从页面缓存直接发送到网络来避免此重新复制。因此在这个优化的路径中，只需要最终拷贝到NIC缓冲区。

我们期望一个常见的用例是多个消费者的主题。使用上面的零拷贝优化，数据被精确地拷贝到页面缓存中一次，并且在每次消费时被重用，而不是被存储在存储器中，并且在每次被读取时被拷贝到内核空间。这允许以接近网络连接限制的速率来消费消息。

这种组合的pagecache和sendfile意味着在Kafka集群上，消费者大多被捕获，你将看不到磁盘上的任何读取活动，因为他们将完全从缓存提供数据。

有关在Java中发送文件和零拷贝支持的详细背景，请参阅本文章。

端到端批量压缩

在某些情况下，瓶颈实际上不是CPU或磁盘，而是网络带宽。这对于需要通过广域网在数据中心之间发送消息的数据管道尤其如此。当然，用户可以总是一次压缩其消息，而不需要来自Kafka的任何支持，但是这可以导致非常差的压缩比，因为大部分冗余是由于相同类型的消息之间的重复（例如，字段名称JSON或Web日志中的用户代理或公共字符串值）。有效压缩需要一起压缩多个消息，而不是单独压缩每个消息。

Kafka通过允许递归消息集来支持这一点。一批消息可以被压缩在一起并以此形式发送到服务器。此批消息将以压缩形式写入，并将在日志中保持压缩，并且只会由消费者解压缩。

Kafka支持GZIP，Snappy和LZ4压缩协议。上压缩更多细节，可以发现在这里。

4.4生产者

负载均衡

生产者直接向作为分区的领导者的代理发送数据，而没有任何中间路由层。为了帮助生产者这样做，所有Kafka节点可以回答关于哪些服务器存活的元数据的请求，以及主题的分区的领导者在任何给定时间处是什么，以允许生产者适当地指示其请求。

客户端控制它发布消息的分区。这可以随机地实现，实现一种随机负载平衡，或者它可以通过一些语义分区功能来完成。我们通过允许用户指定要分区的键并使用它来哈希到分区来暴露语义分区的接口（如果需要，还有一个选项来覆盖分区函数）。例如，如果所选择的密钥是用户ID，则给定用户的所有数据将被发送到同一分区。这反过来将允许消费者对他们的消费做出局部性假设。这种分区风格被明确地设计为允许消费者进行对局部敏感的处理。

异步发送

批处理是效率的主要驱动因素之一，为了启用批处理，Kafka生产者将尝试在内存中累积数据，并在单个请求中发送更大的批次。批处理可以配置为累积不超过固定数量的消息，并且等待不超过一定的固定等待时间（例如64k或10ms）。这允许累积更多的字节发送，并且在服务器上几乎没有更大的I / O操作。这种缓冲是可配置的，并且提供了一种机制来折衷少量的额外延迟以获得更好的吞吐量。

关于详细的配置和API的生产商可以在文档中其他地方找到。

4.5消费者

Kafka消费者通过向经理领导其想要使用的分区发出“提取”请求来工作。消费者在每个请求的日志中指定其偏移量，并从该位置接收回一个日志块。消费者因此对该位置具有重要的控制，并且如果需要，可以将其重绕以重新消费数据。

推送和获取比较

我们考虑的一个初始问题是消费者是否应该从brokers处提取数据或brokers应该将数据推送给消费者。在这方面，Kafka遵循更传统的设计，由大多数消息系统共享，其中数据从生产者推送到代理并由消费者从代理拉出。一些测井为中心的系统，如抄写和 Apache的水槽，按照其中数据下游推一个非常不同的基于推送的路径。两种方法都有利弊。然而，基于推送的系统难以处理各种消费者，因为代理控制数据传送的速率。目标通常是使消费者能够以最大可能的速度消费; 不幸的是，在推送系统中，这意味着当消费者的消费速率低于生产速率（实质上是拒绝服务攻击）时，消费者往往被淹没。基于拉的系统具有更好的性能，消费者简单地落后并且在可能时赶上。这可以通过某种退避协议来减轻，消费者可以通过该协议来指示其被淹没，但是获得传送速率以完全利用（但是从不过度利用）消费者比看起来更麻烦。以前尝试以这种方式建造系统使我们去使用更传统的拉模型。

基于拉的系统的另一个优点是它适合于发送到消费者的数据的积极批处理。基于推送的系统必须选择立即发送请求或累积更多数据，然后稍后发送它，而不知道下游消费者是否能够立即处理它。如果调整为低延迟，这将导致一次发送单个消息，只是为了传输结束被缓冲，这是浪费。基于拉的设计解决了这个问题，因为消费者总是在它在日志中的当前位置（或者最大可配置的最大大小）之后拉出所有可用消息。因此，获得最佳批处理而不引入不必要的延迟。

简单的基于拉的系统的缺陷是，如果代理没有数据，则消费者可能在紧密循环中结束轮询，有效地忙等待数据到达。为了避免这种情况，我们在pull请求中有参数，允许消费者请求在“长轮询”中阻塞，直到数据到达（并且可选地等待直到给定数量的字节可用以确保大的传输大小）。

你可以想象其他可能的设计，只有拉，端到端。生产者在本地写一个本地日志，broker会从消费者拉从他们。通常提出类似类型的“存储 - 转发”生产者。这很有趣，但我们感觉不太适合我们的目标用例有成千上万的生产者。我们在大规模运行持久数据系统方面的经验使我们感到，在许多应用程序中涉及系统中的数千个磁盘实际上不会使事情更可靠，并且是操作的噩梦。在实践中，我们发现我们可以大规模运行具有强SLA的管道，而不需要生产者持久化。

消费者立场

保持跟踪哪些已被消耗是，出人意料的是，一个消息传送系统的关键性能点中的一个。

大多数消息传递系统保存有关在代理上消费的消息的元数据。也就是说，当消息被发送给消费者时，代理或者立即在本地记录该事实，或者它可以等待来自消费者的确认。这是一个相当直观的选择，并且确实对于单个机器服务器，不清楚这个状态在哪里可以去。由于用于许多消息传递系统中的存储的数据结构规模较小，这也是一个务实的选择 - 因为代理知道消耗了什么，它可以立即删除它，保持数据大小小。

可能不明显的是，让broker和消费者对已经消费的东西达成一致并不是一个微不足道的问题。如果代理记录作为消息食用后立刻在网络上流传，每一次，那么如果消费者无法处理的消息（比如，因为它崩溃或请求超时或其他）的信息将会丢失。为了解决这个问题，许多邮件系统添加一个确认特征，这意味着消息仅标记为发送不消耗时它们被发送; 代理等待来自消费者的特定的确认来记录消息作为消耗。此策略解决了丢失邮件的问题，但会产生新的问题。首先，如果消费者在发送确认之前处理消息但失败，则该消息将被消耗两次。第二个问题是性能，现在代理必须保持关于每个单个消息的多个状态（首先锁定它，所以它不是第二次给出，然后标记为永久消费，以便它可以删除）。棘手的问题必须处理，像处理发送但从不确认的消息。

Kafka处理这个不同。我们的主题分为一组完全有序的分区，每个分区在任何给定时间由每个订阅消费者组中的一个消费者消费。这意味着每个分区中的消费者的位置只是一个整数，即要消费的下一条消息的偏移量。这使得关于已经消耗的状态非常小，每个分区只有一个数。此状态可以定期检查点。这使得相当于消息确认非常便宜。

这个决定有一个附带好处。消费者可以故意倒退回老偏移和重新使用数据。这违反了队列的普通合同，但是对于许多消费者来说是一个必要的特征。例如，如果消费者代码具有错误并且在消费一些消息之后被发现，则消费者可以在错误被修复后重新消费这些消息。

离线数据加载

可扩展持久性允许消费者仅仅周期性地消费诸如批量数据加载的消费者的可能性，所述批量数据加载周期性地将数据批量加载到诸如Hadoop或关系数据仓库的离线系统中。

在Hadoop的情况下，我们通过在单个映射任务上分担负载来并行化数据负载，每个节点/主题/分区组合一个，从而允许加载中的完全并行性。Hadoop提供任务管理，失败的任务可以重新启动，没有重复数据的危险 - 他们只是从其原始位置重新启动。

4.6消息传递语义

现在我们了解一下生产者和消费者如何工作，让我们讨论Kafka在生产者和消费者之间提供的语义保证。显然，可以提供多种可能的消息传递保证：

最多一次 -Messages可能会丢失，但从来没有交还。
至少有一次 -Messages从不落空，但可以交还。
恰好一次 -这是人们真正想要的，每个消息被传递一次且仅一次。

值得注意的是，这分为两个问题：发布消息的持久性保证和消息消息的保证。

许多系统声称提供“确切一次”传递语义，但是阅读精细打印是很重要的，这些声明中的大多数是误导的（即，它们不转化为消费者或生产者可能失败的情况，消费者进程，或写入磁盘的数据可能丢失的情况）。

Kafka的语义是直截了当的。当发布消息时，我们有一个消息被“提交”到日志的概念。一旦提交了发布的消息，只要复制写入此消息的分区的一个代理仍保持“活动”，它就不会丢失。下一节将更详细地描述活动的定义以及我们试图处理哪些类型的故障的描述。现在让我们假设一个完美的，无损的broker，并尝试了解对生产者和消费者的保证。如果生产者尝试发布消息并遇到网络错误，则无法确定此错误是在消息提交之前还是之后发生的。这类似于使用自动生成的键插入数据库表的语义。

这些不是发布商最强大的语义。虽然我们不能确定在网络错误的情况下发生了什么，但是可以允许生成器生成一种“主键”，使得重试产生请求幂等。此功能对于复制系统并不重要，因为它必须在服务器故障的情况下工作（或特别）。利用这个特征，生产者重试直到它接收到成功提交的消息的确认就足够了，在该点我们将保证消息已经被精确地发布了一次。我们希望在未来的Kafka版本中添加。

并非所有的用例都需要这样的强大保证。对于对延迟敏感的用途，我们允许生产者指定其期望的耐久性水平。如果生产者指定它希望等待提交的消息，这可以花费10ms的量级。然而，生产者还可以指定它希望完全异步地执行发送，或者它只想等到领导者（但不一定是关注者）具有消息为止。

现在让我们从消费者的角度来描述语义。所有副本具有相同的偏移量的完全相同的日志。消费者控制其在此日志中的位置。如果消费者从未崩溃，它可能只是将这个位置存储在内存中，但是如果消费者失败，并且我们希望这个主题分区被另一个进程占用，新进程将需要选择一个适当的位置开始处理。让我们说消费者读取一些消息 - 它有几个选项来处理消息和更新其位置。

它可以读取消息，然后保存它在日志中的位置，最后处理消息。在这种情况下，消费者进程可能在保存其位置之后但在保存其消息处理的输出之前崩溃。在这种情况下，接管处理的过程将从保存的位置开始，即使该位置之前的一些消息未被处理。这对应于“最多一次”语义，因为在消费者失败消息的情况下可能不被处理。
它可以读取消息，处理消息，最后保存它的位置。在这种情况下，消费者进程可能在处理消息之后但在保存其位置之前崩溃。在这种情况下，当新进程接管它接收的前几个消息将已经被处理。这对应于在消费者失败的情况下的“至少一次”语义。在许多情况下，消息具有主键，因此更新是幂等的（接收相同的消息两次，只是用另一个副本来覆盖记录）。
那么，究竟是什么语义学（即你真正想要的东西）？这里的限制实际上不是消息系统的特征，而是需要协调消费者的位置与实际存储为输出的内容。实现这一点的典型方式是在用于消费者位置的存储器和消费者输出的存储器之间引入两阶段提交。但是这可以通过简单地和简单地通过简单地让消费者将其偏移存储在与其输出相同的位置来处理。这是更好的，因为消费者可能想要写入的许多输出系统将不支持两阶段提交。作为示例，我们的在HDFS中填充数据的Hadoop ETL将其偏移存储在其读取的数据的HDFS中，从而保证数据和偏移都被更新或不被更新。对于需要这些更强的语义并且消息不具有允许重复消除的主键的许多其他数据系统，我们遵循类似的模式。

因此，有效地，Kafka默认保证至少一次传送，并且允许用户通过在生成器上禁用重试并在处理一批消息之前提交其偏移来最多实现一次传送。一次性传送需要与目标存储系统的协作，但Kafka提供了偏移，这使得实现这种直接。

4.7复制

Kafka在可配置的多个服务器上复制每个主题的分区的日志（您可以逐个主题地设置此复制因子）。当集群中的服务器发生故障时，这样可以自动故障转移到这些副本，以便在出现故障时保留可用消息。

其他消息传递系统提供了一些复制相关的功能，但是，在我们的（完全有偏见的）看来，这似乎是一个坚持的事情，没有大量使用，并有大的缺点：奴隶非活动，吞吐量严重影响，手动配置等。Kafka是默认使用复制的 - 事实上，我们将未复制的主题实现为复制主题，其中复制因素为1。

复制单位是主题分区。在非故障情况下，Kafka中的每个分区都有一个领导者和零个或多个关注者。包括前导的复制的总数构成复制因子。所有读取和写入都将转到分区的leader。通常，存在比broker更多的分区，并且领导者被均匀地分布在broker中。跟随者上的日志与领导者的日志相同 - 都具有相同顺序的相同偏移量和消息（当然，在任何给定时间，领导者在其日志末尾可能具有一些未复制的消息）。

关注者像正常的Kafka消费者一样消费领导者的消息，并将它们应用到他们自己的日志中。让追随者从领导者那里拉出来有很好的属性，允许跟随者自然地将他们应用到他们的日志的日志条目排在一起。

与大多数分布式系统一样，自动处理故障需要对节点“活动”意味着什么的精确定义。对于Kafka节点活跃度有两个条件

节点必须能够使用ZooKeeper维护其会话（通过ZooKeeper的心跳机制）
如果它是一个奴隶，它必须复制在领导者上发生的写作，而不是落在“太远”之后

我们将满足这两个条件的节点称为“同步”，以避免“活着”或“失败”的模糊性。领导者跟踪“同步”节点的集合。如果跟随者死亡，被卡住或落后，领导者将从同步副本列表中删除它。滞留和滞后副本的确定由replica.lag.time.max.ms配置控制。

在分布式系统术语中，我们只尝试处理故障的“故障/恢复”模型，其中节点突然停止工作，然后恢复（可能不知道它们已经死亡）。Kafka不处理所谓的“拜占庭式”故障，其中节点产生任意或恶意的响应（可能是由于错误或犯规）。

当同一个分区的所有同步副本都将其应用于其日志时，该消息被认为是“已提交”。只有提交的消息被发送给消费者。这意味着，如果领导失败，消费者不必担心潜在地看到可能丢失的消息。另一方面，生产者可以选择等待消息被提交或不被提交，这取决于他们对延迟和持久性之间权衡的偏好。此偏好由生产者使用的ack设置控制。

Kafka提供的保证是，提交的消息不会丢失，只要至少有一个同步副本活动，在任何时候。

在短期故障转移期后，Kafka将保持可用的节点故障，但在网络分区存在时可能不会保持可用。

复制日志：仲裁，ISR和状态机（Oh my！）

在其核心，Kafka分区是一个复制的日志。复制的日志是分布式数据系统中最基本的原语之一，并且有许多实现方法。已复制日志可以被其他系统中作为一个简单的在实施其他的分布式系统状态机的风格。

复制的日志模拟对一系列值（通常对日志条目0,1,2，...编号）的顺序达成一致的过程。有很多方法来实现这一点，但最简单和最快的是领导者选择提供给它的值的顺序。只要领导者仍然活着，所有追随者都需要仅复制值并且命令领导者选择。

当然，如果领导人没有失败，我们不需要追随者！当领导者死亡时，我们需要从追随者中选择一个新的领导者。但是追随者本身可能落后或崩溃，因此我们必须确保我们选择一个最新的追随者。日志复制算法必须提供的基本保证是，如果我们告诉客户端一个消息被提交，并且领导失败，我们选择的新领导也必须有该消息。这产生了权衡：如果领导等待更多的追随者在声明它提交之前确认消息，则将存在更多潜在的可选领导。

如果选择所需的确认数量和必须比较以选择领导的日志数量，以确保有重叠，则这称为仲裁。

这种权衡的一种常见方法是对提交决策和领导者选举使用多数投票。这不是Kafka做的，但让我们探讨它，以了解权衡。比方说，我们有2 ?F +1副本。如果*?F* +1副本之前必须提交一个由领导宣布收到一条消息，如果我们通过至少拥有最完整的日志选举跟随选出新的领导人 ?F +1副本的话，用不超过*?F故障，领导者保证有所有提交的消息。这是因为任何中间?F* +1副本，必须有一个包含所有提交的信息至少有一个副本。该副本的日志将是最完整的，因此将被选为新的领导者。每个算法必须处理的许多其他细节（例如精确定义什么使日志更完整，确保在领导者失败或更改副本集中的服务器集期间的日志一致性），但我们现在将忽略这些。

这个多数表决方法有一个非常好的属性：延迟只依赖于最快的服务器。也就是说，如果复制因子是三，则延迟由较快的从而不是较慢的一个决定。

有在这个家庭，包括zookeeper的丰富多样的算法朱达，筏和Viewstamped复制。最相似的学术刊物，我们都知道卡夫卡的实际执行情况是 PacificA上来自微软。

多数投票的缺点是，没有什么失败，没有选举领导人离开你。要容忍一个故障，需要三个数据副本，要容忍两个故障，需要五个数据副本。在我们的经验中，只有足够的冗余来容忍单个故障对于实际的系统是不够的，但是每次写入五次，具有5x的磁盘空间要求和1/5吞吐量，对于大容量数据问题不是非常实用。这很可能是为什么仲裁算法更常见的共享集群配置，如ZooKeeper，但不常见的主数据存储。例如，在HDFS的NameNode的高可用性功能是建立在一个基于多数投票杂志，但这种更昂贵的方法不能用于数据本身。

Kafka采用稍微不同的方法来选择其法定集。Kafka不是多数表决，而是动态地维护一组同步副本（ISR），它们被领导者抓住。只有该集合的成员有资格选为领导者。到卡夫卡分区写入，直到不被视为承诺所有在同步副本已接收到写。每当更改时，此ISR集保留到ZooKeeper。因此，ISR中的任何副本都有资格被选为领导者。这是Kafka的使用模型的一个重要因素，其中有很多分区，并确保领导平衡是重要的。有了这个模型ISR和F + 1的复制品，卡夫卡的话题可以容忍*?F*故障不失致力于消息。

对于大多数使用案例，我们希望处理，我们认为这种权衡是一个合理的。在实践中，容忍*?F*倒闭，多数表决和ISR的方法将等待同样数量的副本提交一个消息之前承认两者（如生存一次失败的多数仲裁需要三个副本和一个确认和ISR方法需要两个副本和一个确认）。没有最慢服务器提交的能力是多数表决方法的优势。然而，我们认为通过允许客户端选择是否在消息提交时阻止它们，并且由于所需的复制因子较低而产生的额外吞吐量和磁盘空间是值得的，我们认为它得到改善。

另一个重要的设计区别是Kafka不要求崩溃的节点恢复所有的数据完好无损。在这个空间中的复制算法依赖于“稳定存储”的存在是不常见的，该稳定存储不能在没有潜在一致性违反的任何故障恢复场景中丢失。这个假设有两个主要问题。首先，磁盘错误是我们在持久性数据系统的实际操作中观察到的最常见的问题，并且它们通常不保持数据完整。其次，即使这不是问题，我们不希望在每次写入时使用fsync来保证一致性，因为这可以将性能降低两到三个数量级。我们的允许副本重新加入ISR的协议确保在重新加入之前，它必须再次完全重新同步，即使它在崩溃中丢失未刷新的数据。

不洁的领导选举：如果他们都死了怎么办？

注意，Kafka对数据丢失的保证是基于至少一个保持同步的副本的。如果所有节点复制一个分区死亡，这种保证不再保持。

然而，实际系统需要做一些合理的事情，当所有的副本死。如果你不幸运地发生这种情况，重要的是要考虑会发生什么。有两种可以实现的行为：

等待ISR中的副本恢复生命，并选择这个副本作为领导（希望它仍然有其所有的数据）。
选择第一个复制品（不一定在ISR中）作为领导者回到生活。

这是可用性和一致性之间的简单折衷。如果我们等待ISR中的副本，那么只要这些副本已经关闭，我们将保持不可用。如果这样的副本被破坏或他们的数据丢失，那么我们就永远停止。如果，另一方面，非同步复制品回到生活，我们允许它成为领导者，那么它的日志成为真理的来源，即使它不能保证有每一个提交的消息。默认情况下，Kafka选择第二种策略，并且当ISR中的所有副本都死了时，选择一个潜在不一致的副本。可以使用配置属性unclean.leader.election.enable禁用此行为，以支持停机时间优先于不一致的用例。

这种困境不是Kafka特有的。它存在于任何基于法定人数的方案中。例如，在多数表决方案中，如果大多数服务器遭受永久故障，那么您必须选择丢失100％的数据，或者通过将现有服务器上剩余的数据作为新的真实来源来违反一致性。

可用性和耐久性保证

当写入Kafka时，生产者可以选择它们是等待消息被0,1还是所有（-1）个副本确认。请注意，“所有副本的确认”不保证已分配副本的完整集已接收到该消息。默认情况下，当acks = all时，只要所有当前同步副本都收到消息，就会进行确认。例如，如果一个主题只配置了两个副本并且一个失败（即，只有一个同步副本保留），则指定acks = all的写将成功。但是，如果剩余的副本也失败，这些写入可能会丢失。虽然这确保了分区的最大可用性，但是对于喜欢对可用性的耐久性的一些用户来说，这种行为可能是不期望的。因此，我们提供两个主题级配置，可用于优先考虑消息持久性与可用性：

禁用不干净的领导者选举 - 如果所有副本变得不可用，则分区将保持不可用，直到最近的领导者再次可用。这有效地优选不可用性而不是消息丢失的风险。有关不清洁领导人选举的说明，请参阅上一节。
指定最小ISR大小 - 如果ISR的大小高于某个最小值，则分区将只接受写入，以防止写入到仅一个副本（随后变得不可用）的消息丢失。此设置仅在生产者使用acks = all并保证消息至少由此许多同步副本确认时生效。此设置提供一致性和可用性之间的权衡。最小ISR大小的更高设置确保更好的一致性，因为确保消息被写入更多副本，这降低了它将丢失的可能性。然而，它降低了可用性，因为如果同步副本的数量下降到低于最小阈值，则分区将不可用于写入。

复制管理

上面关于复制日志的讨论实际上只涵盖单个日志，即一个主题分区。然而，Kafka集群将管理成百上千的这些分区。我们尝试以循环方式平衡群集内的分区，以避免在少量节点上为大量主题聚集所有分区。同样，我们尝试平衡领导，以便每个节点是其分区的成比例份额的领导者。

优化领导选举过程也是重要的，因为这是不可用的关键窗口。领导选举的天真实现将最终为每个分区运行一个节点，当该节点发生故障时托管的所有分区。相反，我们选择一个broker作为“控制器”。此控制器检测代理级别的故障，并负责更改故障代理中所有受影响的分区的引导者。结果是，我们能够批准许多所需的领导变更通知，这使得选举过程更简便和更快的大量分区。如果控制器发生故障，其中一个幸存的代理将成为新的控制器。

4.8日志压缩

日志压缩确保Kafka将始终至少保留单个主题分区的数据日志中每个消息键的最后已知值。它解决了应用程序崩溃或系统故障后恢复状态，或在操作维护期间应用程序重新启动后重新加载高速缓存的用例和方案。让我们更详细地深入讨论这些用例，然后描述压缩是如何工作的。

到目前为止，我们仅描述了更简单的数据保留方法，其中在固定时间段之后或当日志达到某个预定大小时丢弃旧日志数据。这对于时间事件数据（例如每个记录独立的记录）很有效。然而，重要的一类数据流是对带密钥的可变数据的更改的日志（例如，对数据库表的更改）。

让我们讨论一个这样的流的具体例子。假设我们有一个主题包含用户电子邮件地址; 每次用户更新他们的电子邮件地址，我们使用他们的用户ID作为主键发送消息到此主题。现在，我们为ID为123的用户在一段时间内发送以下消息，每个消息对应于电子邮件地址的更改（省略其他ID的消息）：

123 => [email protected]
                。
                。
                。
        123 => [email protected]
                。
                。
                。
        123 => [email protected]

登录压实为我们提供了更精细的保留机制，使我们都保证至少保留每个主键（如上次更新

[email protected]

）。通过这样做，我们保证日志包含每个键的最终值的完整快照，而不仅仅是最近更改的键。这意味着下游消费者可以从这个主题中恢复自己的状态，而无需保留所有更改的完整日志。

让我们从看一些有用的用例开始，然后我们将看看如何使用它。

数据库更改订阅。通常需要在多个数据系统中具有数据集，并且这些系统中的一个通常是某种类型的数据库（RDBMS或者新的卷绕的键值存储）。例如，您可能有一个数据库，一个缓存，一个搜索集群和一个Hadoop集群。对数据库的每个更改都需要反映在缓存，搜索集群中，最终在Hadoop中。在一个只处理实时更新的情况下，您只需要最近的日志。但是，如果您想要重新加载缓存或恢复失败的搜索节点，您可能需要一个完整的数据集。
事件采购。这是一种应用程序设计风格，它将查询处理与应用程序设计共存，并将更改日志用作应用程序的主存储。
日记的高可用性。执行本地计算的进程可以通过注销它对其本地状态的更改使容错，以便另一个进程可以重新加载这些更改，如果它应该失败，继续进行。其具体示例是在流查询系统中处理计数，聚合和其他“逐组”类处理。Samza，实时流处理框架，使用此功能出于这样的目的。

在这些情况的每一种情况下，人们都需要主要处理变化的实时馈送，但是偶尔，当机器崩溃或需要重新加载或重新处理数据时，需要做一个满载。日志压缩允许从相同的备份主题提供这两个用例。日志的使用这种风格更详细描述这个博客帖子。

一般的想法很简单。如果我们有无限的日志保留，并且我们记录了上述情况下的每个变化，那么我们将从每次开始时捕获系统的状态。使用这个完整的日志，我们可以通过重播日志中的前N个记录来恢复到任何时间点。这个假设的完整日志对于更新单个记录多次的系统不是非常实用，因为即使对于稳定的数据集，日志也将无限增长。抛弃旧更新的简单日志保留机制将绑定空间，但日志不再是恢复当前状态的方式 - 现在从日志开始恢复不再重新创建当前状态，因为可能不会捕获旧更新。

日志压缩是一种提供更细粒度的每条记录保留的机制，而不是较粗粒度的基于时间的保留。这个想法是选择性地删除具有相同主键的更新的记录。这样，日志保证至少具有每个键的最后状态。

可以为每个主题设置此保留策略，因此单个集群可以具有一些主题，其中通过大小或时间强制保留，以及其他通过压缩实施保留的主题。

此功能是通过LinkedIn的最古老和最成功的作品被称为基础设施数据库更新日志缓存服务的一个启发数据总线。与大多数日志结构存储系统不同，Kafka构建用于订阅和组织数据以进行快速线性读取和写入。与数据库不同，Kafka充当真实来源存储，因此即使在上游数据源不会以其他方式可重放的情况下也是如此。

日志压缩基础

这是一个高级图片，显示每个消息的偏移量的Kafka日志的逻辑结构。

日志的头部与传统的Kafka日志相同。它具有密集的顺序偏移并保留所有消息。日志压缩添加了一个用于处理日志尾部的选项。上图显示了带有紧凑尾巴的日志。请注意，日志尾部中的消息保留在第一次写入时指定的原始偏移量（从不更改）。还要注意，即使具有该偏移量的消息已被压缩，所有偏移仍保留在日志中的有效位置; 在这种情况下，该位置与日志中出现的下一个最高偏移量无法区分。例如，在上图中，偏移36,37和38都是等效位置，并且在这些偏移中的任一个处的读取开始将返回以38开始的消息集。

压缩还允许删除。具有键和空有效内容的邮件将被视为来自日志的删除。此删除标记将导致删除具有该键的任何先前消息（与具有该键的任何新消息一样），但是删除标记是特别的，因为它们将在一段时间之后自己被清除出日志以释放空间。删除不再保留的时间点在上图中标记为“删除保留点”。

压缩在后台通过定期重新复制日志段来完成。清除不会阻止读取，并且可以限制使用不超过可配置的I / O吞吐量，以避免影响生产者和消费者。压缩日志段的实际过程看起来像这样：

日志压缩提供了什么保证？

日志压缩保证以下内容：

任何停留在日志头部内的消费者将看到每个写入的消息; 这些消息将具有顺序偏移。的主题的min.compaction.lag.ms可以用来保证的最小时间长度被写入的消息后它可以被压缩之前，必须通过。即它提供每个消息将保留在（未压缩）头中多长时间的下限。
始终保持消息的排序。压缩永远不会重新排序消息，只是删除一些。
消息的偏移量不会改变。它是日志中位置的永久标识符。
任何一个消费者从日志开始进展将至少看到了最终在他们写的顺序的所有记录状态。对于删除的记录全部删除标记将被视为提供消费者到达日志中的时间周期小于主题的头delete.retention.ms设置（默认为24小时）。这很重要，因为删除标记删除与读取同时发生，因此重要的是，我们不要在消费者看到它之前删除任何删除标记。

日志压缩详细信息

日志压缩由日志清理器处理，日志清理器是一个后台线程池，用于重新复制日志段文件，删除其键出现在日志头中的记录。每个压实机线程工作如下：

它选择日志头到日志尾比率最高的日志
它为日志头中的每个键创建最后一个偏移量的简明摘要
它从开始到结束重新记录日志，删除日志中稍后出现的键。新的，干净的段将立即交换到日志中，因此所需的额外磁盘空间只是一个额外的日志段（而不是日志的完全副本）。
日志头的摘要本质上只是一个空间紧凑的哈希表。它每个条目使用恰好24个字节。因此，使用8GB清洁器缓冲区，一个清洁器迭代可以清除大约366GB的日志头（假设1k个消息）。

配置日志清理器

默认情况下启用日志清除器。这将启动清洁线程池。要对特定主题启用日志清除，可以添加特定于日志的属性

log.cleanup.policy = compact

这可以在主题创建时或使用alter topic命令完成。

日志清理器可以被配置为保留日志的未压缩“头”的最小量。这是通过设置压缩时间延迟启用的。

log.cleaner.min.compaction.lag.ms

这可以用于防止比最小消息时代更新的消息受到压缩。如果未设置，则除了最后一个段（即当前正在写入的段）之外，所有日志段都有资格进行压缩。即使所有的消息都比最小压缩时间滞后更早，活动段也不会被压缩。

进一步的清洁配置描述这里。

4.9配额

从0.9开始，Kafka集群能够对产生和获取请求执行配额。配额基本上是为每组共享配额的客户端定义的字节速率阈值。

为什么需要配额？

生产者和消费者可能生产/消费非常大量的数据，从而独占代理资源，导致网络饱和，并且一般DOS其他客户端和代理本身。具有配额保护以防止这些问题，并且在大型多租户集群中更加重要，其中一小组表现不佳的客户端可以降低良好行为的客户端的用户体验。事实上，当运行Kafka作为服务时，这甚至使得可以根据商定的合同执行API限制。

客户端组

Kafka客户端的身份是表示安全集群中的已认证用户的用户主体。在支持未经身份验证的客户群，用户主要是使用可配置的券商选择了未授权的用户的分组

PrincipalBuilder

。Client-id是客户端应用程序选择的具有有意义名称的客户端的逻辑分组。元组（user，client-id）定义了共享用户主体和客户机标识的客户机的安全逻辑组。

配额可以应用于（用户，客户端标识），用户或客户端ID组。对于给定的连接，将应用与连接匹配的最具体的配额。配额组的所有连接共享为该组配置的配额。例如，如果（user =“test-user”，client-id =“test-client”）的产品配额为10MB /秒，则在用户“test-user”的所有生产者实例之间共享， id“test-client”。

配额配置

可以为（用户，客户端id），用户和客户端id组定义配额配置。可以在需要更高（或甚至更低）配额的任何配额级别覆盖默认配额。该机制类似于per-topic日志配置覆盖。用户和（用户，客户端ID）配额覆盖写入的ZooKeeper下***/config/users和客户端ID的配额被覆盖下写/config/clients***。这些覆盖被所有broker读取，并立即生效。这允许我们更改配额，而不必执行整个群集的滚动重新启动。见这里了解详情。还可以使用相同的机制动态地更新每个组的默认配额。

配额配置的优先级顺序为：

/ config / users / <user> / clients / <client-id>
/ config / users / <user> / clients / <default>
/ config / users / <user>
/ config / users / <default> / clients / <client-id>
/ config / users / <default> / clients / <default>
/ config / users / <default>
/ config / clients / <client-id>
/ config / clients / <default>

代理属性（quota.producer.default，quota.consumer.default）也可用于为客户机标识组设置默认值。这些属性已被弃用，将在以后的版本中删除。可以在Zookeeper中设置client-id的默认配额，类似于其他配额覆盖和默认值。

强制执行

默认情况下，每个唯一的客户端组将按照集群的配置接收固定的配额（以字节/秒为单位）。此配额是基于每个代理定义的。每个客户端可以在其被节制之前每个代理发布/获取最多X字节/秒。我们决定为每个代理定义这些配额比为每个客户端提供固定的群集宽带宽要好得多，因为这需要一种在所有代理之间共享客户端配额使用的机制。这可能比配额实现本身更难得到！

当检测到配额违规时，代理如何做出反应？在我们的解决方案中，代理不返回错误，而是尝试减慢超出其配额的客户端。它计算将有罪客户端置于其配额下所需的延迟量，并延迟该时间的响应。此方法使配额违例对客户端（客户端度量之外）透明。这也使他们不必实施任何特殊的后退和重试行为，可能会变得棘手。事实上，坏的客户端行为（没有后退的重试）可能加剧问题配额正在尝试解决的问题。

客户端字节速率在多个小窗口（例如，每个1秒的30个窗口）上测量，以便快速检测和校正配额违规。通常，具有大的测量窗口（例如，每个30秒的10个窗口）导致大的业务量突发，其后是长的延迟，这在用户体验方面不是很大。

5.实施

5.1 API设计

生产者API

生产者的API封装了2个低级别的生产者-

kafka.producer.SyncProducer

和

kafka.producer.async.AsyncProducer

。

class Producer {

    / *发送数据，按键分隔到主题，使用* /
    / *同步或异步生成器* /
    public void send（kafka.javaapi.producer.ProducerData <K，V> producerData）;

    / *发送数据列表，按键分隔到主题，使用* /
    / *同步或异步生成器* /
    public void send（java.util.List <kafka.javaapi.producer.ProducerData <K，V >> producerData）;

    / *关闭生产者并清理* /
    public void close（）;

    }}

目标是通过单个API向客户端公开所有生产者功能。卡夫卡生产者

可以处理多个生产者请求的排队/缓冲和批量数据的异步分派： kafka.producer.Producer提供的能力，多批次生产的要求（producer.type=async），序列化和将其指派到相应的卡夫卡broker分区之前。批处理的大小可以通过几个配置参数来控制。随着事件的输入队列中，它们被缓冲在队列中，直到queue.time或者batch.size达到。后台线程（kafka.producer.async.ProducerSendThread）会转移到一批数据，并允许kafka.producer.EventHandler序列化和数据发送给相应的卡夫卡broker分区。自定义事件处理程序可通过插在event.handler配置参数。在这个生产者队列管道的各个阶段，能够注入回调是很有帮助的，无论是插入自定义日志/跟踪代码还是自定义监视逻辑。这通过实现是可能kafka.producer.async.CallbackHandler的界面和设置callback.handler配置参数设置为类。
处理数据的序列化通过用户指定的Encoder： interface Encoder<T> { public Message toMessage(T data); } 默认值为no-op kafka.serializer.DefaultEncoder
通过提供软件负载均衡任选用户指定的Partitioner：路由决定由影响kafka.producer.Partitioner。 interface Partitioner<T> { int partition(T key, int numPartitions); } 分区API使用密钥和可用代理分区的数目来返回分区标识。此标识用作对broker_ids和分区的排序列表的索引，以选择生成器请求的代理分区。默认的分区策略是hash(key)%numPartitions。如果密钥为空，则选择随机代理分区。自定义分区策略也可以使用插入partitioner.class配置参数。

消费者API

我们有2个级别的消费者API。低级“简单”API维护与单个代理的连接，并且与发送到服务器的网络请求具有密切的对应关系。这个API是完全无状态的，每个请求都传递偏移量，允许用户维护这些元数据。

高级API隐藏来自消费者的代理的详细信息，并允许在不关心底层拓扑的情况下消耗机器群集。它还维护已消耗的状态。高级API还提供订阅与过滤器表达式匹配的主题（即白名单或黑名单正则表达式）的能力。

低级API

class SimpleConsumer {

    / *向代理发送获取请求，并获取一组消息。* /
    public ByteBufferMessageSet fetch（FetchRequest request）;

    / *向代理发送获取请求的列表，并返回响应集。* /
    public MultiFetchResponse multifetch（List <FetchRequest> fetches）;

    / **
    *在给定时间之前获取有效偏移量（最多maxSize）的列表。
    *结果是偏移列表，按降序排列。
    * @param time：time以毫秒为单位，
    *如果设置为OffsetRequest $ .MODULE $ .LATEST_TIME（），从最新的偏移量获取。
    *如果设置为OffsetRequest $ .MODULE $ .EARLIEST_TIME（），从最早的偏移量获取。
    * /
    public long [] getOffsetsBefore（String topic，int partition，long time，int maxNumOffsets）;
    }}

低级API用于实现高级API，以及直接用于对维护状态有特殊要求的一些离线用户。

高级API

/ *创建到集群的连接* /
    ConsumerConnector connector = Consumer.create（consumerConfig）;

    接口ConsumerConnector {

    / **
    *此方法用于获取KafkaStreams的列表，这是迭代器
    * MessageAndMetadata对象，您可以从中获取消息及其消息
    *相关元数据（目前只有主题）。
    *输入：<topic，#streams>的地图
    *输出：<topic，<消息流列表>的映射
    * /
    public Map <String，List <KafkaStream >> createMessageStreams（Map <String，Int> topicCountMap）;

    / **
    *你也可以获得一个KafkaStreams的列表，它迭代消息
    *来自与TopicFilter匹配的主题。（TopicFilter封装a
    *白名单或作为标准Java正则表达式的黑名单。）
    * /
    public List <KafkaStream> createMessageStreamsByFilter（
        TopicFilter topicFilter，int numStreams）;

    / *提交到目前为止消费的所有消息的偏移量。* /
    public commitOffsets（）

    / *关闭连接器* /
    public shutdown（）
    }}

这个API以迭代器为中心，由KafkaStream类实现。每个KafkaStream表示来自一个或多个服务器上的一个或多个分区的消息流。每个流用于单线程处理，因此客户端可以在创建调用中提供所需流的数量。因此，流可以表示多个服务器分区的合并（以对应于处理线程的数量），但每个分区仅去往一个流。

createMessageStreams调用注册主题的使用者，从而重新平衡使用者/代理分配。API鼓励在单个调用中创建许多主题流，以便最小化此重新平衡。createMessageStreamsByFilter调用（另外）注册观察者以发现与其过滤器匹配的新主题。注意，createMessageStreamsByFilter返回的每个流可以迭代来自多个主题的消息（即，如果过滤器允许多个主题）。

5.2网络层

网络层是相当直接的NIO服务器，并且将不会被详细描述。修正了sendfile实现由给做

MessageSet

接口的

writeTo

方法。这允许设置为使用更有效的文件支持的消息

transferTo

实现，而不是一个进程缓冲的写入。线程模型是一个单一的受主螺纹和Ñ处理器线程其中处理每个连接的一个固定数。这样的设计已经相当彻底的测试在其他地方，发现是实现简单，快速。协议保持相当简单，以允许将来实现其他语言的客户端。

5.3消息

消息由固定大小的报头，可变长度不透明密钥字节数组和可变长度不透明值字节数组组成。标题包含以下字段：

CRC32校验和以检测损坏或截断。
格式版本。
属性标识符
时间戳

离开关键和价值不透明是正确的决定：目前在序列化库上有很大的进步，任何特定的选择不太可能适合所有的用途。不用说，使用Kafka的特定应用程序可能要求特定的序列化类型作为其使用的一部分。该

MessageSet

接口是简单地在与散装读取和写入到NIO专门的方法的消息迭代器

Channel

。

5.4消息格式

/**
        * 1. 4 byte CRC32 of the message
        * 2. 1 byte "magic" identifier to allow format changes, value is 0 or 1
        * 3. 1 byte "attributes" identifier to allow annotations on the message independent of the version
        *    bit 0 ~ 2 : Compression codec.
        *      0 : no compression
        *      1 : gzip
        *      2 : snappy
        *      3 : lz4
        *    bit 3 : Timestamp type
        *      0 : create time
        *      1 : log append time
        *    bit 4 ~ 7 : reserved
        * 4. (Optional) 8 byte timestamp only if "magic" identifier is greater than 0
        * 5. 4 byte key length, containing length K
        * 6. K byte key
        * 7. 4 byte payload length, containing length V
        * 8. V byte payload
        */

5.5 Log

日志名为“my_topic”有两个分区的话题包括两个目录（即

my_topic_0

和

my_topic_1

填充了包含该主题的消息数据文件）。日志文件的格式是“日志条目”，“序列;每个日志条目是一个4字节的整数ñ存储被随后的消息长度ñ消息字节每个消息是唯一由64位整数标识。偏移给出该消息的开始的字节位置在该分区上发送到该主题的所有消息的流中。每个消息的磁盘格式如下：每个日志文件用第一个消息的偏移量命名包含的内容。因此，在第一个文件将00000000000.kafka，并且每个附加文件将有一个整数名大致小号从以前的文件，其中的字节小号是在配置给定的最大日志文件的大小。

消息的确切二进制格式是版本化的，并且保持为标准接口，因此消息集可以在生产者，代理和客户端之间传输，而不需要重新复制或转换。这种格式如下：

消息的磁盘格式

    偏移：8字节 
    消息长度：4字节（值：4 + 1 + 1 + 8（如果magic值> 0）+ 4 + K + 4 +
    crc：4字节
    魔术值：1字节
    属性：1字节
    timestamp：8个字节（只有magic值大于零时才存在）
    密钥长度：4字节
    键：K字节
    值长度：4字节
    值：V字节

使用消息偏移作为消息id是不寻常的。我们最初的想法是使用生产者生成的GUID，并维护从GUID到每个代理上的偏移量的映射。但是由于消费者必须为每个服务器维护一个ID，GUID的全局唯一性不提供任何值。此外，维持从随机ID到偏移的映射的复杂性需要重量权重索引结构，其必须与磁盘同步，基本上需要完全持久随机存取数据结构。因此，为了简化查找??结构，我们决定使用简单的每分区原子计数器，其可以与分区id和节点id耦合以唯一地标识消息; 这使得查找结构更简单，尽管每个消费者请求的多次搜索仍然可能。然而，一旦我们在计数器上定居，直接使用偏移的跳转似乎是自然的 - 都是单调增加分区唯一的整数。由于偏移对消费者API是隐藏的，所以这个决定最终是一个实现细节，我们采用更有效的方法。

写

日志允许串行附加，它总是转到最后一个文件。当文件达到可配置的大小（比如1GB）时，该文件将滚动到新文件。日志采用两个配置参数：中号，这给消息迫使操作系统刷新文件到磁盘之前写入的数目，和小号，这给了若干后冲洗被强制秒。这给顶多失去的耐用性担保中号消息或小号在系统崩溃的事件数据的秒。

阅读

读出由给64位逻辑的消息和一个偏移完成小号 -字节最大块大小。这将返回一个迭代器包含在消息小号 -字节的缓冲区。小号旨在是比任何单一的消息较大，但在异常大的消息的情况下，读出的可以重试多次，每次加倍缓冲区大小，直到所述消息被成功地读取。可以指定最大消息和缓冲区大小，以使服务器拒绝大于某个大小的消息，并给予绑定到客户端的最大值，以便获得完整的消息。很可能读缓冲区以部分消息结束，这很容易通过大小定界来检测。

从偏移量读取的实际过程需要首先定位其中存储数据的日志段文件，从全局偏移值计算文件特定偏移量，然后从该文件偏移量读取。搜索作为对每个文件维护的内存中范围的简单二分搜索变体来完成。

日志提供获取最近写入的消息以允许客户端从“立即”开始订阅的能力。这在消费者未能在其SLA指定的天数内消费其数据的情况下也是有用的。在这种情况下，当客户端尝试使用不存在的偏移量时，它会给出OutOfRangeException，并且可以根据用例重置自身或失败。

以下是发送给消费者的结果的格式。

MessageSetSend（fetch result）

    总长度：4字节
    错误代码：2字节
    消息1：x字节
    ... ...
    消息n：x字节

MultiMessageSetSend（multiFetch result）

    总长度：4字节
    错误代码：2字节
    messageSetSend 1
    ... ...
    messageSetSend n

删除

数据每次删除一个日志段。日志管理器允许可插入的删除策略来选择哪些文件有资格删除。目前的政策会删除所有日志，超过一个修改时间ñ天前，虽然它保留了最后一个政策ñ GB也可能是有用的。为了避免锁定读取，同时仍允许删除修改段列表，我们使用写入时复制样式段列表实现，它提供一致的视图，以允许在删除正在进行时在日志段的不可变静态快照视图上进行二进制搜索。

保证

日志提供配置参数中号控制，强迫冲洗到磁盘之前被写入的消息的最大数量。启动时，将运行日志恢复进程，该进程将迭代最新日志段中的所有消息，并验证每个消息条目是否有效。如果消息条目的大小和偏移的总和小于文件的长度，并且消息有效载荷的CRC32与存储在消息中的CRC匹配，则消息条目是有效的。在检测到损坏的情况下，日志将被截断到最后一个有效偏移量。

注意，必须处理两种损坏：截断，其中未写入块由于崩溃而丢失;以及损坏，其中无意义块被添加到文件。其原因是，一般来说，OS不能保证文件索引节点和实际块数据之间的写入顺序，所以除了丢失写入数据之外，如果索引节点以新的大小更新，则文件可以获得无意义数据，但是崩溃发生在包含该数据的块被写入之前。CRC检测到这种情况，并防止它破坏日志（虽然未写入的消息，当然，丢失）。

5.6分发

消费者偏移跟踪

高级消费者跟踪其在每个分区中消耗的最大偏移，并且周期性地提交其偏移向量，使得其可以在重新启动的情况下从这些偏移重新开始。卡夫卡提供了存储在指定的代理（该组）给定的消费群体所有的偏移称为选项偏移经理。即，该消费者组中的任何消费者实例应将其偏移提交和提取发送到该偏移管理器（代理）。高级消费者自动处理。如果使用简单的消费者，您将需要手动管理偏移量。这在Java简单消费者目前不支持，它只能在ZooKeeper中提交或获取偏移量。如果使用Scala简单使用者，您可以发现偏移管理器，并显式提交或获取偏移到偏移管理器。消费者可以通过向任何Kafka代理发出GroupCoordinatorRequest并读取包含偏移管理器的GroupCoordinatorResponse来查找其偏移管理器。消费者然后可以继续从偏移管理器代理提交或获取偏移。在偏移管理器移动的情况下，消费者将需要重新发现偏移管理器。如果你想手动管理你的偏移量，你可以看看这些解释如何发出OffsetCommitRequest和OffsetFetchRequest代码样本。

当偏移管理器接收到一个OffsetCommitRequest，其附加的请求到一个特殊的压实卡夫卡主题命名*__consumer_offsets*。偏移管理器只有在偏移主题的所有副本都接收到偏移之后才向消费者发送成功的偏移提交响应。如果偏移在可配置的超时内未能复制，则偏移提交将失败，并且消费者可以在后退之后重试提交。（这由高级消费者自动完成。）代理定期压缩偏移主题，因为它只需要维护每个分区的最新偏移提交。偏移管理器还将偏移缓存在内存表中，以便快速地提供偏移提取。

当偏移管理器接收到偏移获取请求时，它简单地从偏移缓存返回最后提交的偏移向量。如果偏移管理器刚刚启动或者它刚刚成为一组新的消费者组的偏移管理器（通过成为偏移主题的分区的领导者），它可能需要将偏移主题分区加载到高速缓存中。在这种情况下，偏移量提取将失败，并出现OffsetsLoadInProgress异常，并且消费者可以在关闭后重试OffsetFetchRequest。（这是由高级消费者自动完成的。）

将偏移量从ZooKeeper迁移到Kafka

早期版本中的Kafka使用者在ZooKeeper中默认存储它们的偏移量。可以通过执行以下步骤来迁移这些消费者以将偏移提交到Kafka：

设置offsets.storage=kafka并dual.commit.enabled=true在您的消费配置。
对消费者进行滚动反弹，然后验证您的消费者是否健康。
设置dual.commit.enabled=false你的消费配置。
对消费者进行滚动反弹，然后验证您的消费者是否健康。

一个回滚（即从卡夫卡回到ZooKeeper的迁移），也可以使用上面的步骤，如果你设置执行

offsets.storage=zookeeper

。

ZooKeeper目录

下面给出了用于消费者和代理之间协调的ZooKeeper结构和算法。

符号

当路径中的元素被表示为[xyz]时，这意味着xyz的值不是固定的，并且实际上存在用于xyz的每个可能值的ZooKeeper znode。例如/ topics / [topic]将是一个名为/ topics的目录，其中包含每个主题名称的子目录。还给出了诸如[0 ... 5]的数字范围以指示子目录0,1,2,3,4。箭头 - >用于指示znode的内容。例如/ hello - > world表示包含值“world”的znode / hello。

代理节点注册表

/“...”，“host”：... /“...” “version”：...，“port”：...}（临时节点）

这是所有当前代理节点的列表，其中每个代理节点提供唯一的逻辑代理ID，用于向消费者标识它（必须作为其配置的一部分）。在启动时，代理节点通过创建具有逻辑代理标识在/ brokers / id下的znode来注册自身。逻辑代理ID的目的是允许将代理移动到不同的物理机，而不影响消费者。尝试注册已在使用的代理标识（因为两个服务器配置为相同的代理标识）导致错误。

由于代理在ZooKeeper中使用临时znode注册自身，所以这种注册是动态的，并且如果代理被关闭或死掉（因此通知消费者它不再可用）它将消失。

代理主题注册表

/brokers/topics/[topic]/partitions/[0...N]/state  - > {“controller_epoch”：...，“leader”：...，“version”：...，“leader_epoch “：...，”isr“：[...]}（临时节点）

每个代理在其维护的主题下注册自己，并存储该主题的分区数。

消费者和消费者群体

主题的消费者也在ZooKeeper中注册自己，以便彼此协调并平衡数据的消耗。消费者还可以通过设置存储在他们的ZooKeeper偏移

offsets.storage=zookeeper

。但是，此偏移量存储机制将在以后的版本中被弃用。因此，建议将偏移存储迁移到卡夫卡。

多个消费者可以形成一个组并共同消费单个主题。同一组中的每个使用者都有一个共享的group_id。例如，如果一个消费者是您的foobar过程，它运行在三台机器上，那么您可以为这组消费者分配id“foobar”。此组ID在消费者的配置中提供，是您告诉消费者其属于哪个组的方式。

组中的消费者尽可能公平地划分分区，每个分区由消费者组中的一个消费者消费。

消费者ID注册表

除了由组中的所有消费者共享的group_id之外，为了识别目的，给每个消费者一个临时的，唯一的consumer_id（形式主机名：uuid）。消费者ID在以下目录中注册。

/ consumers / [group_id] / ids / [consumer_id]  - > {“version”：...，“subscription”：{...：...}，“pattern”：...，“timestamp”： ...}（临时节点）

组中的每个消费者在其组下注册并创建具有其consumer_id的znode。znode的值包含<topic，#streams>的映射。此ID仅用于标识组中当前活动的每个消费者。这是一个临时节点，因此如果消费者进程死亡，它将消失。

消费偏移

消费者跟踪他们在每个分区中消耗的最大偏移量。此值存储在一个zookeeper目录如果

offsets.storage=zookeeper

。

/ consumers / [group_id] / offsets / [topic] / [partition_id]  - > offset_counter_value（persistent node）

分区所有者注册表

每个代理分区由给定用户组中的单个消费者使用。在任何消费可以开始之前，消费者必须建立其对给定分区的所有权。为了建立它的所有权，消费者在它声称的特定代理分区下的临时节点中写入它自己的id。

/ consumers / [group_id] / owners / [topic] / [partition_id]  - > consumer_node_id（ephemeral node）

集群ID

集群ID是分配给Kafka集群的唯一且不可变的标识符。集群ID最多可以有22个字符，允许的字符由正则表达式[a-zA-Z0-9 _ \ - ] +定义，它对应于没有填充的URL安全Base64变体使用的字符。从概念上讲，当集群第一次启动时，它是自动生成的。

在实现方面，它是第一次成功启动版本0.10.1或更高版本的代理时生成的。该券商试图获取从集群ID

/cluster/id

启动期间znode。如果znode不存在，则代理生成新的集群标识，并创建具有此集群标识的znode。

代理节点注册

代理节点基本上是独立的，因此它们只发布关于它们具有的信息。当代理加入时，它会在代理节点注册表目录下注册自己，并写入有关其主机名和端口的信息。代理还在代理主题注册表中注册现有主题及其逻辑分区的列表。在代理上创建新主题时，会动态地注册新主题。

消费者注册算法

当消费者启动时，它执行以下操作：

在其组下的consumer id注册表中注册自己。
在消费者id注册表下注册更改（新消费者加入或任何现有消费者离开）。（每个更改触发在更改的消费者所属的组内的所有消费者之间的重新平衡。）
在broker id注册表下注册一个更改（新broker加入或任何现有的broker离开）。（每个更改触发所有客户组中所有客户之间的重新平衡。）
如果消费者使用主题过滤器创建消息流，它还会在代理主题注册表下注册对更改（正在添加的新主题）的监视。（每个更改将触发对可用主题的重新评估，以确定主题过滤器允许哪些主题）。新的允许主题将触发消费者组中所有消费者之间的重新平衡。
强制自己在消费群体内重新平衡。

消费者重新平衡算法

消费者重新平衡算法允许组中的所有消费者对哪个消费者消费哪些分区达成共识。每次添加或删除同一组中的代理节点和其他使用者时，会触发消费者重新平衡。对于给定主题和给定使用者组，代理分区在组内的用户间平均分配。分区总是由单个消费者使用。这种设计简化了实现。如果我们允许一个分区被多个消费者同时使用，则在分区上存在争用并且将需要某种类型的锁定。如果消费者比分区多，一些消费者根本不会获得任何数据。在重新平衡期间，我们尝试为消费者分配分区，以减少每个消费者必须连接到的代理节点的数量。

每个消费者在重新平衡期间执行以下操作：

1.对于C中的话题牛逼我赞成
    2.设P 牛逼是生产主题T总分区
    3.令C 摹是所有消费者在同一组为C 我消耗的话题?
    4.排序P 牛逼（依此类推同一代理分区聚集在一起）
    5.排序ç 摹 
    6 i为C的索引位置我用C 摹，让N =大小（P 牛逼）/尺寸（C 摹）
    7.指定分区从i * N至第（i + 1）* N - 1对消费者ç 我 
    8.取出用C拥有当前条目我从分区所有者注册表
    9.将新分配的分区添加到分区所有者注册表
            （我们可能需要重新尝试，直到原始分区所有者释放其所有权）

当在一个消费者处触发重新平衡时，应当在相同时间内在相同组内的其他消费者中触发重新平衡。

6.操作

下面是一些关于实际运行Kafka作为基于在LinkedIn的使用和经验的生产系统的信息。请将您知道的任何其他提示发送给我们。

6.1基本Kafka操作

本部分将回顾您将在Kafka集群上执行的最常见操作。所有在本节回顾了工具下可

bin/

卡夫卡分布目录，如果不带参数运行每一个工具将在所有可能的命令行选项打印的详细信息。

添加和删除主题

您可以选择手动添加主题，或在首次将数据发布到不存在的主题时自动创建主题。如果主题是自动创建的，那么你可能需要调整默认的主题配置用于自动创建的主题。

使用主题工具添加和修改主题：

> bin/kafka-topics.sh --zookeeper zk_host:port/chroot --create --topic my_topic_name
        --partitions 20 --replication-factor 3 --config x = y

复制因素控制有多少服务器将复制每个写入的消息。如果复制因子为3，则最多2个服务器可能会失败，然后您将无法访问数据。我们建议您使用2或3的复制因子，以便可以透明地反弹机器，而不会中断数据消耗。

分区计数控制主题将被分成多少日志。分区计数有几个影响。首先每个分区必须完全适合单个服务器。因此，如果您有20个分区，则完全数据集（以及读写负载）将由不超过20个服务器（不计算副本）处理。最后，分区计数会影响消费者的最大并行性。这是在更详细地讨论的概念部分。

每个分片分区日志都放在Kafka日志目录下的自己的文件夹中。这些文件夹的名称由主题名称（用短划线（ - ）和分区ID附加）组成。由于典型的文件夹名称不能超过255个字符长，因此对主题名称的长度将有限制。我们假设分区的数量不会超过100,000。因此，主题名称不能超过249个字符。这留下足够的空间在文件夹名称中的破折号和可能的5位长的分区标识。

在命令行中添加的配置将覆盖服务器对于应保留数据的时间长度的默认设置。一套完整的每个主题的配置被记录在这里。

修改主题

您可以使用相同的主题工具更改主题的配置或分区。

要添加分区，您可以做

> bin/kafka-topics.sh --zookeeper zk_host:port/chroot --alter --topic my_topic_name
        - 分区40

请注意，分区的一个用例是对数据进行语义分区，并且添加分区不会更改现有数据的分区，因此如果这些数据依赖于该分区，则可能会干扰消费者。也就是说，如果数据被划分

hash(key) % number_of_partitions

那么这个分区将可能被打乱通过添加分区，但卡夫卡不会试图以任何方式自动重新分配数据。

添加配置：

> bin/kafka-topics.sh --zookeeper zk_host:port/chroot --alter --topic my_topic_name --config x = y

删除配置：

> bin/kafka-topics.sh --zookeeper zk_host:port/chroot --alter --topic my_topic_name --delete-config x

最后删除主题：

> bin/kafka-topics.sh --zookeeper zk_host：port/chroot --delete --topic my_topic_name

默认情况下禁用主题删除选项。要启用它设置服务器配置

delete.topic.enable = true

Kafka当前不支持减少主题的分区数。

改变一个主题的复制因子的说明可以发现在这里。

正常关机

Kafka集群将自动检测任何代理关闭或失败，并为该计算机上的分区选择新的领导者。无论服务器发生故障还是故意关闭维护或配置更改，都会发生这种情况。对于后一种情况，Kafka支持一种更优雅的停止服务器的机制，而不仅仅是杀死它。当服务器正常停止时，它有两个优化，它将利用：

它会将所有日志同步到磁盘，以避免在重新启动时（即验证日志尾部的所有消息的校验和）进行任何日志恢复。日志恢复需要时间，以便加速故意重新启动。
它将在关闭之前将服务器所在的任何分区迁移到其他副本。这将使领导传输更快，并将每个分区不可用的时间减少到几毫秒。

只要服务器停止而不是硬杀死，同步日志将自动发生，但受控领导迁移需要使用特殊设置：

controlled.shutdown.enable = true

注意：如果控制停机才会成功都托管在券商的分区有副本（即复制因子大于1 并且这些副本中的至少一个还活着）。这通常是你想要的，因为关闭最后一个副本会使该主题分区不可用。

平衡领导

每当代理停止或崩溃领导，代理的分区转移到其他副本。这意味着默认情况下，当代理重新启动时，它将只是所有其分区的跟随器，这意味着它不会用于客户端读取和写入。

为了避免这种不平衡，Kafka有一个首选副本的概念。如果分区的副本的列表为1,5,9，则节点1优选为节点5或9的引导者，因为它在副本列表中较早。您可以通过运行以下命令让Kafka集群尝试恢复恢复的副本的领导：

> bin/kafka-preferred-replica-election.sh --zookeeper zk_host:port/chroot

由于运行此命令可能很繁琐，您还可以配置Kafka通过设置以下配置自动执行此操作：

auto.leader.rebalance.enable = true

平衡机架间的副本

机架感知功能将相同分区的副本分布在不同机架上。这扩展了Kafka提供的用于代理故障以保证机架故障的保证，如果机架上的所有代理同时发生故障，则限制数据丢失的风险。该功能还可以应用于其他代理分组，如EC2中的可用区域。

您可以通过向代理配置添加属性来指定代理属于特定机架：

broker.rack = my-rack-id

当一个主题创建，修改或复制品被重新分配，机架约束将兑现，确保副本跨越尽可能多架，因为他们可以（分区将跨越分钟（#racks，复制因子）不同的机架）。

用于向代理分配副本的算法确保每个代理的领导数量将是不变的，而不管代理是如何分布在机架上的。这确保平衡吞吐量。

但是，如果机架被分配不同数量的代理，则副本的分配将不均匀。具有较少代理的机架将获得更多的副本，这意味着它们将使用更多的存储，并将更多的资源用于复制。因此，每个机架配置相等数量的代理是明智的。

在集群之间镜像数据

我们指的是复制数据的过程之间的卡夫卡集群“镜像”，以避免这种情况发生在一个群集中的节点之间复制混乱。Kafka自带了一个用于在Kafka集群之间镜像数据的工具。该工具从源群集消耗并生成到目标群集。这种镜像的常见用例是在另一个数据中心中提供副本。这种情况将在下一节中更详细地讨论。

您可以运行许多这样的镜像过程以增加吞吐量和容错（如果一个进程死机，其他进程将占用额外的负载）。

将从源群集中的主题读取数据，并将其写入目标群集中具有相同名称的主题。事实上，镜子制造商只是一个卡夫卡消费者和制造商钩在一起。

源和目标集群是完全独立的实体：它们可以具有不同数量的分区，并且偏移将不相同。由于这个原因，镜像集群不是真正意图作为容错机制（因为消费者位置将不同）; 因此我们建议使用正常的群集内复制。然而，镜像制造者进程将保留并使用消息键来进行分区，所以在每个键的基础上保存订单。

这里是展示如何反映一个主题一个例子（名为MY-话题从输入集群）：

> bin/kafka-mirror-maker.sh
        --consumer.config consumer.properties
        --producer.config producer.properties --whitelist my-topic

请注意，我们指定的主题列表中

--whitelist

选择。此选项允许使用任何正则表达式Java风格的正则表达式。所以，你可以叫镜像两个话题一个和乙使用

--whitelist 'A|B'

。或者你可以镜像所有使用的主题

--whitelist '*'

。确保引用任何正则表达式以确保shell不会尝试将其作为文件路径展开。为了方便起见，我们允许使用'，'而不是'|' 以指定主题列表。

有时它是容易说什么是你不想要的。而不是使用

--whitelist

说要镜像你可以用什么

--blacklist

该说什么排除在外。这也需要一个正则表达式参数。然而，

--blacklist

当新的消费已被启用，不支持（即当

bootstrap.servers

已经在消费者配置中定义）。

结合配置镜像

auto.create.topics.enable=true

使得它有一个副本群集会自动创建和复制，即使在添加新的课题源集群中的所有数据的可能。

检查消费者的位置

有时，查看您的消费者的位置很有用。我们有一个工具，它将显示消费者组中所有消费者的位置，以及它们在日志结尾处有多远。要运行在一个消费群体这个工具叫我组耗费了话题叫我，话题是这样的：

> bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 --group test
  Group           Topic                          Pid Offset          logSize         Lag             Owner
  my-group        my-topic                       0   0               0               0               test_jkreps-mn-1394154511599-60744496-0
  my-group        my-topic                       1   0               0               0               test_jkreps-mn-1394154521217-1a0be913-0

注意：自0.9.0.0以来，kafka.tools.ConsumerOffsetChecker工具已被弃用。您应该使用kafka.admin.ConsumerGroupCommand（或bin/kafka-consumer-groups.sh脚本）来管理消费群体，包括与消费者建立新的消费API。

管理用户组

使用ConsumerGroupCommand工具，我们可以列出，描述或删除用户组。请注意，删除仅在组元数据存储在ZooKeeper中时可用。当使用新的消费API（其中broker处理的分区处理和重新平衡协调），该集团是当该组到期最后提交的偏移删除。例如，要列出所有主题中的所有用户组：

> bin/kafka-consumer-groups.sh --bootstrap-server broker1:9092 --list

  test-consumer-group

要使用ConsumerOffsetChecker查看前一个示例中的偏移量，我们按如下方式“描述”使用者组：

> bin/kafka-consumer-groups.sh --bootstrap-server broker1:9092 --describe --group test-consumer-group

  GROUP                          TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER
  test-consumer-group            test-foo                       0          1               3               2               consumer-1_/127.0.0.1

如果您使用的是旧的高层次消费者和存储在zookeeper（即组元

offsets.storage=zookeeper

），通

--zookeeper

而不是

bootstrap-server

：

> bin / kafka-consumer-groups.sh --zookeeper localhost：2181 --list

扩展群集

将服务器添加到Kafka集群很容易，只需为它们分配一个唯一的代理ID，并在新服务器上启动Kafka。但是，这些新服务器不会自动分配任何数据分区，因此除非将分区移动到这些分区，否则在创建新主题之前，它们不会进行任何工作。因此，通常当您将计算机添加到集群时，您将要将某些现有数据迁移到这些计算机。

迁移数据的过程是手动启动的，但完全自动化。下面是发生的情况是，Kafka将添加新的服务器作为其正在迁移的分区的跟随者，并允许它完全复制该分区中的现有数据。当新服务器完全复制此分区的内容并加入同步副本时，现有副本之一将删除其分区的数据。

分区重新分配工具可用于跨分组移动分区。理想的分区分布将确保跨所有代理的均匀的数据加载和分区大小。分区重新分配工具不具有自动研究Kafka集群中的数据分布并移动分区以获得均匀负载分布的能力。因此，管理员必须确定应该移动哪些主题或分区。

分区重新分配工具可以以3种互斥模式运行：

--generate：在此模式下，给定主题列表和代理列表，该工具生成候选重新??分配，以将指定主题的所有分区移动到新代理。此选项仅提供一个方便的方法来生成分区重新分配计划给定的主题和目标代理的列表。
--execute：在此模式下，工具将根据用户提供的重新分配计划启动分区的重新分配。（使用--reassignment-json-file选项）。这可以是由管理员手工制作的自定义重新分配计划，也可以使用--generate选项提供
--verify：在此模式下，该工具将验证在最后执行过程中列出的所有分区的重新分配状态。状态可以是成功完成，失败或正在进行

自动将数据迁移到新计算机

分区重新分配工具可以用于将一些主题从当前代理集合移动到新添加的代理。这在扩展现有集群时通常很有用，因为更容易将整个主题移动到新的代理集合，而不是一次移动一个分区。当用于这样做时，用户应该提供应当移动到新的代理集合的主题列表和新的代理的目标列表。然后，该工具在新的代理集合中均匀分配给定主题列表的所有分区。在此移动期间，主题的复制因子保持不变。有效地，主题输入列表的所有分区的副本从旧的代理集合移动到新添加的代理。

例如，以下示例将主题foo1，foo2的所有分区移动到新的代理集合5,6。在这一举措的结束，对于主题foo1和foo2的所有分区将只在broker5,6存在。

由于工具接受主题的输入列表作为json文件，您首先需要确定要移动的主题，并创建json文件如下：

> cat themes-to-move.json
  {“topics”：[{“topic”：“foo1”}，
              {“topic”：“foo2”}]，
  “version”：1
  }}

一旦json文件准备就绪，使用分区重新分配工具生成候选分配：

> bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --topics-to-move-json-file topic-to-move.json --broker-list“5,6”--generate
  当前分区副本分配

  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：2，“replicas”：[1,2]}，
                {“topic”：“foo1”，“partition”：0，“replicas”：[3,4]}，
                {“topic”：“foo2”，“partition”：2，“replicas”：[1,2]}，
                {“topic”：“foo2”，“partition”：0，“replicas”：[3,4]}，
                {“topic”：“foo1”，“partition”：1，“replicas”：[2,3]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[2,3]}]
  }}

  建议的分区重新分配配置

  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：2，“replicas”：[5,6]}，
                {“topic”：“foo1”，“partition”：0，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：2，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：0，“replicas”：[5,6]}，
                {“topic”：“foo1”，“partition”：1，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[5,6]}]
  }}

该工具生成一个候选分配，将所有分区从主题foo1，foo2移动到代理5,6。但请注意，在这一点上，分区移动尚未开始，它只是告诉你当前分配和建议的新分配。应保存当前分配以防万一要回滚到它。新的分配应该保存在一个json文件（例如expand-cluster-reassignment.json）中，使用--execute选项输入到工具中，如下所示：

> bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file expand-cluster-reassignment.json --execute
  当前分区副本分配

  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：2，“replicas”：[1,2]}，
                {“topic”：“foo1”，“partition”：0，“replicas”：[3,4]}，
                {“topic”：“foo2”，“partition”：2，“replicas”：[1,2]}，
                {“topic”：“foo2”，“partition”：0，“replicas”：[3,4]}，
                {“topic”：“foo1”，“partition”：1，“replicas”：[2,3]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[2,3]}]
  }}

  将其保存为在回滚期间用作--reassignment-json-file选项
  已成功启动重新分配分区
  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：2，“replicas”：[5,6]}，
                {“topic”：“foo1”，“partition”：0，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：2，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：0，“replicas”：[5,6]}，
                {“topic”：“foo1”，“partition”：1，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[5,6]}]
  }}

最后，--verify选项可以与工具一起使用以检查分区重新分配的状态。请注意，与-verify选项一起使用相同的expand-cluster-reassignment.json（与--execute选项一起使用）

> bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file expand-cluster-reassignment.json --verify
  分区重新分配的状态：
  重新分配分区[foo1,0]已成功完成
  正在重新分配分区[foo1,1]
  正在重新分配分区[foo1,2]
  重新分配分区[foo2,0]已成功完成
  重新分配分区[foo2,1]已成功完成
  重新分配分区[foo2,2]已成功完成

自定义分区分配和迁移

分区重新分配工具还可以用于选择性地将分区的副本移动到特定的代理集合。当以这种方式使用时，假定用户知道重新分配计划并且不需要工具生成候选重新??分配，有效地跳过 - 生成步骤并且直接移动到执行步骤

例如，以下示例将主题foo1的分区0移动到代理5,6，将主题foo2的分区1移动到代理2,3：

第一步是在json文件中手工制作自定义重分配计划：

> cat custom-reassignment.json
  {“version”：1，“partitions”：[{“topic”：“foo1”，“partition”：0，“replicas”：[5,6]}，{“topic”：“foo2” ：1，“replicas”：[2,3]}]}

然后，使用带有--execute选项的json文件来启动重新分配过程：

> bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file custom-reassignment.json --execute
  当前分区副本分配

  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：0，“replicas”：[1,2]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[3,4]}]
  }}

  将其保存为在回滚期间用作--reassignment-json-file选项
  已成功启动重新分配分区
  {“version”：1，
  “partitions”：[{“topic”：“foo1”，“partition”：0，“replicas”：[5,6]}，
                {“topic”：“foo2”，“partition”：1，“replicas”：[2,3]}]
  }}

--verify选项可以与工具一起使用以检查分区重新分配的状态。请注意，与-verify选项一起使用相同的expand-cluster-reassignment.json（与--execute选项一起使用）

bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file custom-reassignment.json --verify
  分区重新分配的状态：
  重新分配分区[foo1,0]已成功完成
  重新分配分区[foo2,1]已成功完成

退役broker

分区重新分配工具无法自动为已停用的代理自动生成重新分配计划。因此，管理员必须提出重新分配计划，以将代理上托管的所有分区的副本移动到其余代理。这可能是相对冗长的，因为重新分配需要确保所有副本不会从已退役的代理移动到仅一个其他代理。为了使这个过程轻松，我们计划在未来为退役broker增加工具支持。

增加复制因子

增加现有分区的复制因素很容易。只需在自定义重新分配json文件中指定额外的副本，并与--execute选项一起使用以增加指定分区的复制因子。

例如，以下示例将主题foo的分区0的复制因子从1增加到3.在增加复制因子之前，分区的唯一副本存在于代理5上。作为增加复制因子的一部分，我们将添加更多副本经纪6和7。

第一步是在json文件中手工制作自定义重分配计划：

> cat increase-replication-factor.json
  {“version”：1，
  “partitions”：[{“topic”：“foo”，“partition”：0，“replicas”：[5,6,7]}]

然后，使用带有--execute选项的json文件来启动重新分配过程：

> bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file increase-replication-factor.json --execute
  当前分区副本分配

  {“version”：1，
  “partitions”：[{“topic”：“foo”，“partition”：0，“replicas”：[5]}]}

  将其保存为在回滚期间用作--reassignment-json-file选项
  已成功启动重新分配分区
  {“version”：1，
  “partitions”：[{“topic”：“foo”，“partition”：0，“replicas”：[5,6,7]}]

--verify选项可以与工具一起使用以检查分区重新分配的状态。请注意，与--verify选项一起使用相同的increase-replication-factor.json（与--execute选项一起使用）

bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --reassignment-json-file increase-replication-factor.json --verify
  分区重新分配的状态：
  重新分配分区[foo，0]已成功完成

您还可以使用kafka-topics工具验证复制因子的增加：

> bin / kafka-topics.sh --zookeeper localhost：2181 --topic foo --describe
  主题：foo PartitionCount：1 ReplicationFactor：3配置：
    主题：foo分区：0领导：5副本：5,6,7 Isr：5,6,7

限制数据迁移期间的带宽使用

Kafka允许您对复制流量应用节流，设置用于将副本从机器移动到机器的带宽的上限。在重新平衡集群，引导新代理或添加或删除代理时，这是非常有用的，因为它限制了这些数据密集型操作对用户的影响。

有两个接口可用于接合节气门。最简单，最安全的是在调用kafka-reassign-partitions.sh时应用节流阀，但是kafka-configs.sh也可以用于直接查看和更改节流值。

因此，例如，如果您要执行重新平衡，使用以下命令，它将移动分区不超过50MB / s。

$ bin / kafka-reassign-partitions.sh --zookeeper myhost：2181  -  execute --reassignment-json-file greater-cluster.json -throttle 50000000

当你执行这个脚本你会看到油门接合：

节流极限设置为50000000B / s
  已成功启动重新分配分区。

如果你想改变油门，在重新平衡期间，说增加吞吐量，以便完成更快，你可以通过重新运行execute命令通过相同的reassignment-json文件：

$ bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --execute --reassignment-json-file greater-cluster.json --throttle 700000000
  有一个现有的分配正在运行。
  节气门限设置为700000000 B / s

一旦重新平衡完成，管理员可以使用--verify选项检查重新平衡的状态。如果重新平衡已完成，油门将通过--verify命令删除。重要的是，一旦重新平衡完成，管理员通过运行带有--verify选项的命令及时删除油门。否则可能会导致定期复制流量受到限制。

当执行--verify选项并且重新分配完成时，脚本将确认油门已被删除：

$ bin / kafka-reassign-partitions.sh --zookeeper localhost：2181 --verify --reassignment-json-file bigger-cluster.json
  分区重新分配的状态：
  重新分配分区[my-topic，1]已成功完成
  重新分配分区[mytopic，0]已成功完成
  油门被拆除。

管理员还可以使用kafka-configs.sh验证分配的配置。有两对油门配置用于管理节流过程。油门值本身。这是在代理级使用动态属性配置的

还有一组枚举的复制副本：

leader.replication.throttled.replicas
  follower.replication.throttled.replicas

其中每个主题配置。所有四个配置值由kafka-reassign-partitions.sh（下面讨论）自动分配。

查看油门限制配置：

$ bin / kafka-configs.sh --describe --zookeeper localhost：2181 --entity-type brokers
  Brokers'2'的配置为leader.replication.throttled.rate = 700000000，follower.replication.throttled.rate = 700000000
  代理'1'的配置为leader.replication.throttled.rate = 700000000，follower.replication.throttled.rate = 700000000

这显示应用于复制协议的前导和跟随侧的节流阀。默认情况下，两侧都分配有相同的节制吞吐量值。

要查看受限复制副本的列表，请执行以下操作：

$ bin / kafka-configs.sh --describe --zookeeper localhost：2181 --entity-type topics
  主题'my-topic'的配置为leader.replication.throttled.replicas = 1：102,0：101，
      follower.replication.throttled.replicas = 1：101,0：102

这里我们看到领导节流被应用于代理102上的分区1和代理101上的分区0.同样地，追随者节流被应用于代理101上的分区1和代理102上的分区0。

默认情况下，kafka-reassign-partitions.sh会将领导节气门应用于重新平衡前存在的所有副本，任何一个副本都可能是leader。它将应用随动油门到所有移动目的地。因此，如果存在具有在代理101,102上的副本的分区，被重新分配给102,103，则用于该分区的领导节流将被应用于101,102，并且追随者节流将仅被应用于103。

如果需要，您还可以使用kafka-configs.sh上的--alter开关手动更改油门配置。

安全使用节流复制

在使用节流复制时应该格外小心。尤其是：

（1）油门拆卸：

一旦重新分配完成（通过运行kafka-reassign-partitions -verify），应及时删除节流阀。

（2）确保进展：

如果油门设置得太低，与传入的写入速率相比，复制可能无法进行。这发生在：

max（BytesInPerSec）> throttle

其中BytesInPerSec是监视生产者到每个代理中的写入吞吐量的度量。

管理员可以使用度量来监控重新平衡期间复制是否正在进行：

kafka.server：type = FetcherLagMetrics，name = ConsumerLag，clientId =（[ - 。\ w] +），topic =（[ - 。\ w] +），partition =（[0-9] +）

在复制期间滞后应该不断减少。如果度量不减小，则管理员应该如上所述增加节流吞吐量。

设置配额

配额覆盖和默认值可以在（用户，客户端ID），用户或客户端ID水平描述的配置在这里。默认情况下，客户端接收无限额的配额。可以为每个（用户，客户端标识），用户或客户端标识组设置自定义配额。

为（user = user1，client-id = clientA）配置自定义配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type users --entity-name user1  - -entity-name clientA
  更新实体的配置：user-principal'user1'，client-id'clientA'。

为user = user1配置自定义配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type users --entity-name user1
  更新实体的配置：user-principal'user1'。

为client-id = clientA配置自定义配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type clients --entity-name clientA
  更新实体的配置：client-id'clientA'。

它可以通过指定设置每个（用户，客户端ID）的默认配额，用户或客户端ID组*--entity默认选项，而不--entity名*。

为user = userA配置默认客户端标识配额。

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type users --entity-name user1  - -entity-default
  更新实体的配置：user-principal'user1'，默认客户端标识。

配置用户的默认配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type users --entity-default
  更新实体的配置：default user-principal。

为客户端标识配置默认配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --alter --add-config'producer_byte_rate = 1024，consumer_byte_rate = 2048'--entity-type clients --entity-default
  更新实体的配置：default client-id。

以下说明如何描述指定（使用者，用户端ID）的配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --describe --entity-type users --entity-name user1 --entity-type clients --entity-name clientA
  用户主体'user1'，client-id'clientA'的配置为producer_byte_rate = 1024，consumer_byte_rate = 2048

描述给定用户的配额：

描述给定客户端ID的配额：

> bin / kafka-configs.sh --zookeeper localhost：2181 --describe --entity-type clients --entity-name clientA
  客户端ID“clientA”的配置为producer_byte_rate = 1024，consumer_byte_rate = 2048

如果未指定实体名称，则将描述指定类型的所有实体。例如，描述所有用户：

> bin / kafka-configs.sh --zookeeper localhost：2181 --describe --entity-type users
  用户主体'user1'的配置为producer_byte_rate = 1024，consumer_byte_rate = 2048
  默认用户主体的配置为producer_byte_rate = 1024，consumer_byte_rate = 2048

类似的（用户，客户端）：

> bin / kafka-configs.sh --zookeeper localhost：2181 --describe --entity-type users --entity-type clients
  配置用户主体'user1'，默认客户端标识为producer_byte_rate = 1024，consumer_byte_rate = 2048
  用户主体'user1'，client-id'clientA'的配置为producer_byte_rate = 1024，consumer_byte_rate = 2048

可以通过在代理上设置这些配置来设置适用于所有客户端标识的默认配额。仅当未在Zookeeper中配置配额覆盖或缺省值时，才应用这些属性。默认情况下，每个客户端标识接收无限制的配额。以下将每个生产者和消费者客户端标识的默认配额设置为10MB /秒。

quota.producer.default = 10485760
    quota.consumer.default = 10485760

请注意，这些属性已被弃用，可能会在将来的版本中删除。使用kafka-configs.sh配置的默认值优先于这些属性。

6.2数据中心

一些部署将需要管理跨多个数据中心的数据管道。我们推荐的方法，这是部署在应用程序实例每个数据中心本地卡夫卡集群中的每个数据中心只与本地集群的互动和集群之间的镜像（请参见文档镜制造工具，如何做到这一点）。

此部署模式允许数据中心充当独立实体，并允许我们集中管理和调整数据中心间复制。这允许每个设施独立和操作，即使数据中心间链路不可用：当这发生时，镜像落后，直到链路被恢复，在该时间它赶上。

对于需要可以使用镜像提供具有从本地集群镜像汇总数据集群的所有数据的全局视图，应用程序的所有数据中心。这些聚集集群用于需要完整数据集的应用程序的读取。

这不是唯一可能的部署模式。可以通过WAN从远程Kafka集群读取或写入，虽然显然这将增加获得集群所需的任何延迟。

Kafka自然地在生产者和消费者中分批数据，因此即使在高延迟的连接上也可以实现高吞吐量。允许此虽然它可能有必要增加使用的生产者，消费者和broker TCP套接字缓冲区大小

socket.send.buffer.bytes

和

socket.receive.buffer.bytes

配置。适当的方法来设置这是记录在这里。

它一般是不建议运行一个单一的跨越高延迟链接多个数据中心卡夫卡集群。这将为Kafka写入和ZooKeeper写入带来非常高的复制延迟，如果位置之间的网络不可用，Kafka和ZooKeeper都不会在所有位置保持可用。

6.3 Kafka配置

重要的客户端配置

最重要的生产者配置控制

压缩
同步与异步生产
批量大小（用于异步生成器）

最重要的消费者配置是获取大小。

所有配置都记录在配置部分。

生产服务器配置

这里是我们的生产服务器配置：

＃复制配置
  num.replica.fetchers = 4
  replica.fetch.max.bytes = 1048576
  replica.fetch.wait.max.ms = 500
  replica.high.watermark.checkpoint.interval.ms = 5000
  replica.socket.timeout.ms = 30000
  replica.socket.receive.buffer.bytes = 65536
  replica.lag.time.max.ms = 10000

  controller.socket.timeout.ms = 30000
  controller.message.queue.size = 10

  ＃日志配置
  num.partitions = 8
  message.max.bytes = 1000000
  auto.create.topics.enable = true
  log.index.interval.bytes = 4096
  log.index.size.max.bytes = 10485760
  log.retention.hours = 168
  log.flush.interval.ms = 10000
  log.flush.interval.messages = 20000
  log.flush.scheduler.interval.ms = 2000
  log.roll.hours = 168
  log.retention.check.interval.ms = 300000
  log.segment.bytes = 1073741824

  ＃ZK配置
  zookeeper.connection.timeout.ms = 6000
  zookeeper.sync.time.ms = 2000

  ＃套接字服务器配置
  num.io.threads = 8
  num.network.threads = 8
  socket.request.max.bytes = 104857600
  socket.receive.buffer.bytes = 1048576
  socket.send.buffer.bytes = 1048576
  queued.max.requests = 16
  fetch.purgatory.purge.interval.requests = 100
  producer.purgatory.purge.interval.requests = 100

我们的客户端配置在不同的用例之间存在相当大的差异。

6.4 Java版本

从安全角度来看，我们建议您使用JDK 1.8的最新发布版本，因为较旧的免费版本已经公开了安全漏洞。LinkedIn目前正在使用G1收集器运行JDK 1.8 u5（希望升级到更新的版本）。如果您决定使用G1收集器（当前默认值），并且您仍然使用JDK 1.7，请确保您使用的是u51或更高版本。LinkedIn在测试中尝试u21，但是他们在该版本中有一些问题。LinkedIn的调整方式如下：

-Xmx6g -Xms6g -XX：MetaspaceSize = 96m -XX：+ UseG1GC
  -XX：MaxGCPauseMillis = 20 -XX：InitiatingHeapOccupancyPercent = 35 -XX：G1HeapRegionSize = 16M
  -XX：MinMetaspaceFreeRatio = 50 -XX：MaxMetaspaceFreeRatio = 80

作为参考，以下是LinkedIn的最繁忙的集群之一（峰值）的统计信息：

60 broker
50k分区（复制因子2）
800k讯息/秒
300 MB /秒入站，1 GB /秒+出站

调整看起来相当积极，但是该集群中的所有代理有90％的GC暂停时间约21ms，他们每秒做少于1个年轻的GC。

6.5硬件和操作系统

我们使用具有24GB内存的双核四核Intel Xeon机器。

您需要足够的内存来缓冲活动的读写器和写入器。您可以通过假设您希望能够缓冲30秒，并将您的内存需求计算为write_throughput * 30，对内存需求进行后备估计。

磁盘吞吐量很重要。我们有8x7200 rpm SATA驱动器。一般来说，磁盘吞吐量是性能瓶颈，并且更多的磁盘更好。根据您如何配置刷新行为，您可能或可能不会从更昂贵的磁盘（如果强制刷新经常，然后更高RPM SAS驱动器可能更好）。

操作系统

Kafka应该在任何unix系统上运行良好，并已在Linux和Solaris上测试。

我们已经看到在Windows和Windows上运行的几个问题目前不是一个很好的支持平台，但我们很乐意改变。

它不太可能需要大量的操作系统级调整，但有两个潜在的重要的操作系统级配置：

文件描述符限制：Kafka对日志段和打开的连接使用文件描述符。如果代理主持许多分区，那么考虑代理需要至少（number_of_partitions）*（partition_size / segment_size）来跟踪所有日志段以及代理做出的连接数。我们建议至少有100000个允许的文件描述符作为代理进程的起点。
最大套接字缓冲区大小：可以增加以使数据中心之间的高性能数据传输这里描述。

磁盘和文件系统

我们建议使用多个驱动器以获得良好的吞吐量，而不是与应用程序日志或其他操作系统文件系统活动共享用于Kafka数据的相同驱动器，以确保良好的延迟。您可以将这些驱动器一起RAID成一个卷或格式，并将每个驱动器作为其自己的目录。由于Kafka具有复制功能，RAID提供的冗余也可以在应用程序级别提供。这个选择有几个折衷。

如果配置多个数据目录，分区将被轮询分配给数据目录。每个分区将完全在一个数据目录中。如果数据在分区之间没有很好地平衡，这可能导致磁盘之间的负载不平衡。

RAID可以在平衡磁盘之间的负载方面做得更好（虽然它并不总是这样），因为它在较低级别上平衡负载。RAID的主要缺点是它通常对写吞吐量造成很大的性能损失，并减少可用磁盘空间。

RAID的另一个潜在好处是能够容忍磁盘故障。然而，我们的经验是，重建RAID阵列是如此I / O密集型，它有效地禁用服务器，因此这不提供很多实际可用性改进。

应用程序与操作系统刷新管理

Kafka总是立即将所有数据写入文件系统，并支持配置刷新策略的能力，该策略控制数据何时被强制从操作系统缓存强制推出并使用刷新进入磁盘。可以控制该刷新策略以在一段时间之后或在写入了一定数量的消息之后将数据强制到磁盘。在此配置中有几个选择。

Kafka最终必须调用fsync知道数据被刷新。当从崩溃中恢复任何未知为fsync的日志段时，Kafka将通过检查每个消息的CRC来检查每个消息的完整性，并且还将重新生成伴随的偏移索引文件，作为启动时执行的恢复过程的一部分。

请注意，Kafka中的持久性不需要将数据同步到磁盘，因为失败的节点将始终从其副本中恢复。

我们建议使用完全禁用应用程序fsync的默认刷新设置。这意味着依赖于操作系统和Kafka自己的后台刷新完成的后台刷新。这为大多数用途提供了所有世界中最好的：没有旋钮调整，大吞吐量和延迟，以及完全恢复保证。我们通常认为复制提供的保证比同步到本地磁盘更强，然而偏执仍然可能更喜欢同时支持和应用程序级fsync策略。

使用应用程序级别刷新设置的缺点是它的磁盘使用模式效率较低（它给操作系统减少了重新排序写操作的余地），并且可能引入延迟，因为fsync在大多数Linux文件系统中阻止写入文件，而后台刷新进行更细粒度的页面级锁定。

一般来说，您不需要对文件系统进行任何低级调优，但在接下来的几个部分中，我们将介绍一些有用的文件系统。

了解Linux操作系统刷新行为

在Linux中，写入到文件系统的数据保持在页缓存，直到它必须被写入到磁盘（由于应用程序级FSYNC或操作系统自身的冲洗政策）。数据的刷新是通过一组称为pdflush（或在2.6.32内核“刷新线程”）的后台线程完成的。

Pdflush有一个可配置的策略，控制在缓存中维护多少脏数据，以及在它必须写回磁盘之前多长时间。这一政策被描述这里。当Pdflush不能跟上写入数据的速率时，它将最终导致写入过程阻止写入中的延迟，以减慢数据的累积。

您可以通过执行查看操作系统内存使用的当前状态

> cat / proc / meminfo

这些值的含义在上面的链接中描述。

使用pagecache比用于存储要写入磁盘的数据的进程内高速缓存有几个优点：

I / O调度器将连续的小写入批处理成更大的物理写入，这提高了吞吐量。
I / O调度器将尝试重新排序写入以最小化磁头的移动，这提高了吞吐量。
它自动使用机器上的所有可用内存

文件系统选择

Kafka在磁盘上使用常规文件，因此它对特定文件系统没有硬依赖性。但是，使用最多的两个文件系统是EXT4和XFS。历史上，EXT4已经有更多的使用，但最近对XFS文件系统的改进表明，它对Kafka的工作负载具有更好的性能特性，而不会影响稳定性。

使用各种文件系统创建和装载选项对具有显着消息负载的集群执行比较测试。被监控的Kafka中的主要度量标准是“请求本地时间”，指示附加操作占用的时间量。XFS导致更好的本地时间（160ms对于最好的EXT4配置的250ms +），以及较低的平均等待时间。XFS性能也显示磁盘性能的变化性较小。

一般文件系统说明

对于用于数据目录的任何文件系统，在Linux系统上，建议在安装时使用以下选项：

noatime：当读取文件时，此选项禁用更新文件的atime（最后访问时间）属性。这可以消除大量的文件系统写入，特别是在自举消费者的情况下。Kafka不依赖atime属性，所以可以安全地禁用它。

XFS注释

XFS文件系统具有大量的自动调整功能，因此无需在文件系统创建时或挂载时对默认设置进行任何更改。值得考虑的唯一调整参数是：

largeio：这影响stat调用报告的首选I / O大小。虽然这可以允许在较大的磁盘写入上实现更高的性能，但在实践中它对性能的影响极小或没有影响。
nobarrier：对于具有电池备份缓存的底层设备，此选项通过禁用定期写入刷新可以提供一些更多的性能。但是，如果底层设备运行良好，它会向文件系统报告它不需要刷新，并且此选项将不起作用。

EXT4注释

EXT4是用于Kafka数据目录的文件系统的可服务选择，然而获得最佳性能将需要调整几个装载选项。此外，这些选项在故障情况下通常是不安全的，并且将导致更多的数据丢失和损坏。对于单个代理故障，这不是很关心，因为可以擦除磁盘，并从群集重建副本。在多故障情况下，如断电，这可能意味着底层文件系统（因此数据）损坏，这是不容易恢复的。可以调整以下选项：

data = writeback：Ext4默认为data = ordered，对一些写入设置强顺序。Kafka不需要这个顺序，因为它对所有未刷新的日志执行非常偏执的数据恢复。此设置删除了排序约束，并且似乎显着减少延迟。
禁用日志记录：日志记录是一种权衡：它使服务器崩溃后重新启动更快，但它引入了大量额外的锁定，这增加了写入性能的方差。那些不关心重启时间并且想减少写入延迟峰值的主要来源的用户可以完全关闭日志功能。
commit = num_secs：这将调整ext4提交到其元数据日志的频率。将此值设置为较低值可减少崩溃期间未刷新数据的丢失。将其设置为更高的值将提高吞吐量。
nobh：此设置在使用data = writeback模式时控制额外的排序保证。这应该是安全的与Kafka，因为我们不依赖于写顺序，并提高吞吐量和延迟。
delalloc：延迟分配意味着文件系统避免分配任何块，直到物理写入发生。这允许ext4分配大范围而不是较小的页面，并有助于确保数据顺序写入。此功能对于吞吐量非常重要。它似乎涉及在文件系统中增加一点延迟差异的一些锁定。

6.6监测

Kafka使用Yammer度量在服务器和客户端中的度量报告。这可以配置为使用可插入统计记录器报告统计信息，以连接到监控系统。

查看可用度量标准的最简单方法是启动jconsole并将其指向正在运行的kafka客户端或服务器; 这将允许浏览所有度量与JMX。

我们对以下指标进行图形化和警报：

描述

MBEAN名称

正常值

留言率

kafka.server：type = BrokerTopicMetrics，name = MessagesInPerSec

字节速率

kafka.server：type = BrokerTopicMetrics，name = BytesInPerSec

请求速率

kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower}

字节输出速率

kafka.server：type = BrokerTopicMetrics，name = BytesOutPerSec

日志刷新率和时间

kafka.log：type = LogFlushStats，name = LogFlushRateAndTimeMs

＃of under replicated partitions（| ISR | <| all replicas |）

kafka.server：type = ReplicaManager，name = underReplicatedPartitions

代理上控制器处于活动状态

kafka.controller：type = KafkaController，name = ActiveControllerCount

在集群中只有一个代理应该有1

领导选举率

kafka.controller：type = ControllerStats，name = LeaderElectionRateAndTimeMs

非零，当有代理失败

不洁的领导选举率

kafka.controller：type = ControllerStats，name = UncleanLeaderElectionsPerSec

分区计数

kafka.server：type = ReplicaManager，name = PartitionCount

甚至跨broker

领导者副本计数

kafka.server：type = ReplicaManager，name = LeaderCount

甚至跨broker

ISR收缩率

kafka.server：type = ReplicaManager，name = IsrShrinksPerSec

如果代理断开，某些分区的ISR将收缩。当代理重新启动时，一旦副本完全被占用，ISR将被扩展。除此之外，ISR收缩率和扩张率的预期值为0。

ISR扩展率

kafka.server：type = ReplicaManager，name = IsrExpandsPerSec

往上看

消息中的最大滞后btw跟随者和领导者副本

kafka.server：type = ReplicaFetcherManager，name = MaxLag，clientId = Replica

滞后应与产品请求的最大批量大小成比例。

每个追随者副本的邮件滞后

kafka.server：type = FetcherLagMetrics，name = ConsumerLag，clientId =（[ - 。\ w] +），topic =（[ - 。\ w] +），partition =（[0-9] +）

滞后应与产品请求的最大批量大小成比例。

请求在生产者炼狱中等待

kafka.server：type = DelayedOperationPurgatory，name = PurgatorySize，delayedOperation = Produce

非零，如果使用ack = -1

请求在获取缓冲区中等待

kafka.server：type = DelayedOperationPurgatory，name = PurgatorySize，delayedOperation = Fetch

大小取决于消费者中的fetch.wait.max.ms

请求总时间

kafka.network:type=RequestMetrics,name=TotalTimeMs,request={Produce|FetchConsumer|FetchFollower}

分成队列，本地，远程和响应发送时间

请求在请求队列中等待的时间

kafka.network:type=RequestMetrics,name=RequestQueueTimeMs,request={Produce|FetchConsumer|FetchFollower}

在领导处理请求的时间

kafka.network:type=RequestMetrics,name=LocalTimeMs,request={Produce|FetchConsumer|FetchFollower}

请求等待跟随者的时间

kafka.network:type=RequestMetrics,name=RemoteTimeMs,request={Produce|FetchConsumer|FetchFollower}

当ack = -1时，非零用于产生请求

请求在响应队列中等待的时间

kafka.network:type=RequestMetrics,name=ResponseQueueTimeMs,request={Produce|FetchConsumer|FetchFollower}

发送响应的时间

kafka.network:type=RequestMetrics,name=ResponseSendTimeMs,request={Produce|FetchConsumer|FetchFollower}

消费者滞后于生产者的消息数

kafka.consumer：type = ConsumerFetcherManager，name = MaxLag，clientId =（[ - 。\ w] +）

网络处理器空闲的平均时间分数

kafka.network:type=SocketServer,name=NetworkProcessorAvgIdlePercent

在0和1之间，理想地> 0.3

请求处理程序线程空闲的平均时间部分

kafka.server：type = KafkaRequestHandlerPool，name = RequestHandlerAvgIdlePercent

在0和1之间，理想地> 0.3

每（用户，客户端ID），用户或客户端ID的配额指标

kafka.server：type = {Produce | Fetch}，user =（[ - 。\ w] +），client-id =（[ - 。\ w] +）

两个属性。throttle-time表示客户端被限制的时间（以毫秒为单位）。理想值为0.字节速率表示客户端的数据产生/消耗速率（以字节/秒为单位）。对于（user，client-id）配额，指定user和client-id。如果将per-client-id配额应用于客户端，则不指定用户。如果应用每用户配额，则不指定client-id。

生产者/消费者/连接的共同监控指标

以下度量可用于生产者/消费者/连接器实例。有关具体指标，请参阅以下部分。

指标/属性名称

描述

MBEAN名称

连接关闭率

窗口中每秒关闭的连接。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

连接创建率

窗口中每秒建立的新连接。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

网络速率

每秒所有连接上的平均网络操作数（读取或写入）。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

传出字节率

每秒向所有服务器发送的传出字节的平均数。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

请求率

每秒发送的平均请求数。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

request-size-avg

窗口中所有请求的平均大小。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

request-size-max

窗口中发送的任何请求的最大大小。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

传入字节率

字节/秒读取所有插座。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

反应速度

收到每秒发送的响应。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

选择率

I / O层每秒检查新I / O执行的次数。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

io-wait-time-ns-avg

I / O线程花费在等待以纳秒为单位准备好读取或写入的套接字的平均时间长度。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

io等待比

I / O线程花费等待的时间比例。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

io-time-ns-avg

每个选择呼叫的I / O的平均时间长度（以纳秒为单位）。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

io比

I / O线程用于执行I / O的时间比例。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

连接计数

当前活动连接数。

kafka。[producer | consumer | connect]：type = [producer | consumer | connect] -metrics，client-id =（[ - 。\ w] +）

生产者/消费者/连接的公共每个代理指标

以下度量可用于生产者/消费者/连接器实例。有关具体指标，请参阅以下部分。

指标/属性名称

描述

MBEAN名称

传出字节率

每个节点每秒发送的传出字节的平均数。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

请求率

每个节点每秒发送的平均请求数。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

request-size-avg

窗口中所有请求的平均大小。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

request-size-max

在窗口中为节点发送的任何请求的最大大小。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

传入字节率

每个节点每秒接收的平均响应数。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

request-latency-avg

节点的平均请求延迟（以毫秒为单位）。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

request-latency-max

节点的最大请求延迟（以毫秒为单位）。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

反应速度

每秒为节点发送的响应数。

kafka.producer：type = [consumer | producer | connect] -node-metrics，client-id =（[ - 。\ w] +），node-id =（[0-9] +）

生产者监控

以下度量标准可用于生产者实例。

指标/属性名称

描述

MBEAN名称

等待线程

用户线程数阻塞等待缓冲区内存使其记录入队。