阿里大牛的Kafka动态配置了解下

在开始分享之前，我们先来复习一下设置 Kafka 参数，特别是 Broker 端参数的方法。

在 Kafka 安装目录的 config 路径下，有个 serverproperties 文件。通常情况下，我们会指定这个文件的路径来启动 Broker。如果要设置 Broker 端的任何参数，我们必须在这个文件中显式地增加一行对应的配置，之后启动 Broker 进程，令参数生效。我们常见的做法是，一次性设置好所有参数之后，再启动 Broker。当后面需要变更任何参数时，我们必须重启 Broker。但生产环境中的服务器，怎么能随意重启呢？所以，目前修改 Broker 端参数是非常痛苦的过程。

基于这个痛点，社区于 110 版本中正式引入了动态 Broker 参数（Dynamic BrokerConfigs）。所谓动态，就是指修改参数值后，无需重启 Broker 就能立即生效，而之前在serverproperties 中配置的参数则称为静态参数（Static Configs）。显然，动态调整参数值而无需重启服务，是非常实用的功能。如果你想体验动态 Broker 参数的话，那就赶快升级到 11 版本吧。

当然了，当前最新的 23 版本中的 Broker 端参数有 200 多个，社区并没有将每个参数都升级成动态参数，它仅仅是把一部分参数变成了可动态调整。那么，我们应该如何分辨哪些参数是动态参数呢？

如果你打开 11 版本之后（含 11）的 Kafka 官网，你会发现Broker Configs表中增加了Dynamic Update Mode 列。该列有 3 类值，分别是 read-only、per-broker 和 cluster-wide。我来解释一下它们的含义。

我来举个例子说明一下 per-broker 和 cluster-wide 的区别。Broker 端参数 listeners 想必你应该不陌生吧。它是一个 per-broker 参数，这表示你只能为单个 Broker 动态调整listeners，而不能直接调整一批 Broker 的 listeners。logretentionms 参数是 cluster-wide 级别的，Kafka 允许为集群内所有 Broker 统一设置一个日志留存时间值。当然了，你也可以为单个 Broker 修改此值。

你可能会问，动态 Broker 参数的使用场景都有哪些呢？实际上，因为不必重启 Broker，动态 Broker 参数的使用场景非常广泛，通常包括但不限于以下几种：

在这些使用场景中，动态调整线程池大小应该算是最实用的功能了。很多时候，当 KafkaBroker 入站流量（inbound data）激增时，会造成 Broker 端请求积压（Backlog）。有了动态参数，我们就能够动态增加网络线程数和 I/O 线程数，快速消耗一些积压。当突发流量过去后，我们也能将线程数调整回来，减少对资源的浪费。整个过程都不需要重启Broker。你甚至可以将这套调整线程数的动作，封装进定时任务中，以实现自动扩缩容。

由于动态配置的特殊性，它必然有和普通只读参数不同的保存机制。下面我来介绍一下Kafka 是如何保存动态配置的。

首先，Kafka 将动态 Broker 参数保存在 ZooKeeper 中，具体的 znode 路径如下图所示。

我来解释一下图中的内容。changes 是用来实时监测动态参数变更的，不会保存参数值；topics 是用来保存 Kafka 主题级别参数的。虽然它们不属于动态 Broker 端参数，但其实它们也是能够动态变更的。

users 和 clients 则是用于动态调整客户端配额（Quota）的 znode 节点。所谓配额，是指Kafka 运维人员限制连入集群的客户端的吞吐量或者是限定它们使用的 CPU 资源。

分析到这里，我们就会发现，/config/brokers znode 才是真正保存动态 Broker 参数的地方。该 znode 下有两大类子节点。第一类子节点就只有一个，它有个固定的名字叫 <default >，保存的是前面说过的 cluster-wide 范围的动态参数；另一类则以 brokerid 为名，保存的是特定 Broker 的 per-broker 范围参数。由于是 per-broker 范围，因此这类子节点可能存在多个。

我们一起来看一张，它展示的是我的一个 Kafka 集群环境上的动态 Broker 端参数。

在这张图中，我首先查看了 /config/brokers 下的子节点，我们可以看到，这里面有 <default > 节点和名为 0、1 的子节点。< default > 节点中保存了我设置的 cluster-wide范围参数；0 和 1 节点中分别保存了我为 Broker 0 和 Broker1 设置的 per-broker 参数。

接下来，我分别展示了 cluster-wide 范围和 per-broker 范围的参数设置。拿numiothreads 参数为例，其 cluster-wide 值被动态调整为 12，而在 Broker 0 上被设置成 16，在 Broker 1 上被设置成 8。我为 Broker 0 和 Broker 1 单独设置的值，会覆盖掉cluster-wide 值，但在其他 Broker 上，该参数默认值还是按 12 计算。

如果我们再把静态参数加进来一起讨论的话，cluster-wide、per-broker 和 static 参数的优先级是这样的：per-broker 参数 > cluster-wide 参数 > static 参数 > Kafka 默认值。

另外，如果你仔细查看上图中的 ephemeralOwner 字段 ，你会发现它们的值都是 0x0。这表示这些 znode 都是持久化节点，它们将一直存在。即使 ZooKeeper 集群重启，这些数据也不会丢失，这样就能保证这些动态参数的值会一直生效。

讲完了保存原理，我们来说说如何配置动态 Broker 参数。目前，设置动态参数的工具行命令只有一个，那就是 Kafka 自带的 kafka-configs 脚本。接下来，我来以uncleanleaderelectionenable 参数为例，演示一下如何动态调整。

下面这条命令展示了如何在集群层面设置全局值，即设置 cluster-wide 范围值。

总体来说命令很简单，但有一点需要注意。 如果要设置 cluster-wide 范围的动态参数，需要显式指定 entity-default 。现在，我们使用下面的命令来查看一下刚才的配置是否成功。

从输出来看，我们成功地在全局层面上设置该参数值为 true。注意 sensitive=false 的字眼，它表明我们要调整的参数不是敏感数据。如果我们调整的是类似于密码这样的参数时，该字段就会为 true，表示这属于敏感数据。

好了，调整完 cluster-wide 范围的参数，我来演示下如何设置 per-broker 范围参数。我们还是以 uncleanleaderelectionenable 参数为例，我现在为 ID 为 1 的 Broker 设置一个不同的值。命令如下：

同样，我们使用下列命令，来查看一下刚刚的设置是否生效了。

这条命令的输出信息很多。我们关注两点即可。

如果我们要删除 cluster-wide 范围参数或 per-broker 范围参数，也非常简单，分别执行下面的命令就可以了。

删除动态参数要指定 delete-config 。当我们删除完动态参数配置后，再次运行查看命令，结果如下：

此时，刚才配置的所有动态参数都已经被成功移除了。

刚刚我只是举了一个参数的例子，如果你想要知道动态 Broker 参数都有哪些，一种方式是在 Kafka 官网中查看 Broker 端参数列表，另一种方式是直接运行无参数的 kafka-configs脚本，该脚本的说明文档会告诉你当前动态 Broker 参数都有哪些。我们可以先来看看下面这两张图。

看到有这么多动态 Broker 参数，你可能会问：这些我都需要调整吗？你能告诉我最常用的几个吗？根据我的实际使用经验，我来跟你分享一些有较大几率被动态调整值的参数。

1logretentionms 。

修改日志留存时间应该算是一个比较高频的 *** 作，毕竟，我们不可能完美地预估所有业务的消息留存时长。虽然该参数有对应的主题级别参数可以设置，但拥有在全局层面上动态变更的能力，依然是一个很好的功能亮点。

2numiothreads 和 numnetworkthreads 。

这是我们在前面提到的两组线程池。就我个人而言，我觉得这是动态 Broker 参数最实用的场景了。毕竟，在实际生产环境中，Broker 端请求处理能力经常要按需扩容。如果没有动态 Broker 参数，我们是无法做到这一点的。

3 与 SSL 相关的参数 。

主要是 4 个参数（sslkeystoretype、sslkeystorelocation、sslkeystorepassword 和sslkeypassword）。允许动态实时调整它们之后，我们就能创建那些过期时间很短的 SSL证书。每当我们调整时，Kafka 底层会重新配置 Socket 连接通道并更新 Keystore。新的连接会使用新的 Keystore，阶段性地调整这组参数，有利于增加安全性。

4numreplicafetchers 。

这也是我认为的最实用的动态 Broker 参数之一。Follower 副本拉取速度慢，在线上Kafka 环境中一直是一个老大难的问题。针对这个问题，常见的做法是增加该参数值，确保有充足的线程可以执行 Follower 副本向 Leader 副本的拉取。现在有了动态参数，你不需要再重启 Broker，就能立即在 Follower 端生效，因此我说这是很实用的应用场景。

本文重点讨论了 Kafka 110 版本引入的动态 Broker 参数。这类参数最大的好处在于，无需重启 Broker，就可以令变更生效，因此能够极大地降低运维成本。除此之外，还给出了动态参数的保存机制和设置方法。

kafka的常用命令

新建一个topic

/bin/kafka-topicssh --create --zookeeper hadoop04:2181,hadoop05:2181,hadoop06:2181 --partition 3 --replication-factor 3 --topic test-0

test-0是topic的名字

replication-factor 3 副本数

topic存储的是元数据，通过它找到真实的数据

改变分区

/bin/kafka-topicssh --alter --zookeeper 19216814131:2181,19216814131:2182,19216814131:2183 --partition 5 --topic test03

分区信息查询

/bin/kafka-topicssh --describe --zookeeper hadoop04:2181,hadoop05:2181,hadoop06:2181 --topic test-0

结果为：leader主分区（broker Id） replicationfactor副本数 Isr是否存活（存储的顺序）

Topic:test-0 PartitionCount:3 ReplicationFactor:3 Configs:

Topic: test-0 Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1

Topic: test-0 Partition: 1 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2

Topic: test-0 Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

查询语句：

bin/kafka-topicssh --list --zookeeper hadoop04:2181,hadoop05:2181,hadoop06:2181

删除语句

bin/kafka-topicssh --delete --zookeeper hadoop04:2181,hadoop05:2181,hadoop06:2181 --topic test-0

结果是test-0 - marked for deletion

表示不能真正的删除只是做了个标记要删除取zookeeper去真正删除（这个配置如果设置为true可以真正删除，设置false不能删除）

#删除topic需要serverproperties中设置deletetopicenable=true否则只是标记删除

deletetopicenable=false

rmr /brokers/topics/test-0 ，删除client中的brokers中topic对应的

rmr /config/topics/test-0 ,删配置

rmr /admin/delete_topics，删除admin中的标记为删除的topics

分区设置后只可以增加不可以减少

六API *** 作

启动生产者

/bin/kafka-console-producersh --broker-list 19216814131:9092,19216814132:9092,19216814133:9092 --topic test03

启动消费者

/bin/kafka-console-consumersh --zookeeper 192168247131:2181,192168247132:2181,192168247133:2181 --topic test03 --from-beginning

Kafka 的命令行工具在 Kafka 包的/bin目录下，主要包括服务和集群管理脚本，配置脚本，信息查看脚本，Topic 脚本，客户端脚本等。

kafka-configssh：配置管理脚本

kafka-console-consumersh：kafka 消费者控制台

kafka-console-producersh：kafka 生产者控制台

kafka-consumer-groupssh：kafka 消费者组相关信息

kafka-delete-recordssh：删除低水位的日志文件

kafka-log-dirssh：kafka 消息日志目录信息

kafka-mirror-makersh：不同数据中心 kafka 集群复制工具

kafka-preferred-replica-electionsh：触发 preferred replica 选举

kafka-producer-perf-testsh：kafka 生产者性能测试脚本

kafka-reassign-partitionssh：分区重分配脚本

kafka-replica-verificationsh：复制进度验证脚本

kafka-server-startsh：启动 kafka 服务

kafka-server-stopsh：停止 kafka 服务

kafka-topicssh：topic 管理脚本

kafka-verifiable-consumersh：可检验的 kafka 消费者

kafka-verifiable-producersh：可检验的 kafka 生产者

zookeeper-server-startsh：启动 zk 服务

zookeeper-server-stopsh：停止 zk 服务

zookeeper-shellsh：zk 客户端

我们通常可以使用kafka-console-consumersh和kafka-console-producersh脚本来测试 Kafka 生产和消费，kafka-consumer-groupssh可以查看和管理集群中的 Topic，kafka-topicssh通常用于查看 Kafka 的消费组情况。

本文将介绍最常用的分布式消息中间件kafka。由于笔者水平受限，因此介绍不一定全面，也不会太深入，仅供参考。

如果启动时提示命令语法不正确，那么需要在kafka安装目录中找到bin\windows目录中的kafka-run-classbat，为set COMMAND后面的%CLASSPATH%加上双引号

注意22版本可以直接用--bootstrap-server替代--zookeeper

一条消息是一个record batch，包含record batch header，每条record又有各自的header

一个segment由index和log组成。index是索引文件，记录每条消息的offset和在log中的地址，log中存储具体的数据。segment大小固定，但是包含不同数目的消息，segment文件的命名由上一个segment的最后一条消息的offset决定。查询指定offset消息的过程是先通过二分查找找到对应的segment，然后在index文件中通过二分查找找到对应的存储地址。

compaction指对相同key的数据进行合并。

每个partition都有一个leader，若干个followers，读写请求发送给leader处理。leader维护了一个isr（in-sync replicas）列表，写数据时只有当指定数量的isr告知已收到（acknowledge）leader才会commit，而数据只有commit之后才会被消费者看到。告知已收到的数量可以由producer决定，包括0,1或者all（-1）

如果分区的当前leader挂掉了，会从isr列表中重新选举leader。如果列表中的所有节点都挂掉了，那么有以下几种策略

以  zookeeper的地址：19216815041      端口：2185     为例

hosts记录：   19216815041   zk41

1连接zoo：

bin/zookeeper-shellsh     19216815041:2185       或者      bin/zookeeper-shellsh     zk41:2185 

简单几个命令：

1ls

2查看broker信息：

get /brokers/ids/0

常用命令：

进入服务器后，找到kafka安装目录进入bin文件夹，输入命令---

查看kafka现有主题命令:。/kafka-topicssh --list --zookeeper zk_host:port

在启动Kafka之前，需要启动zookeeper，否则会报错！相关的启动指令如下：

在此配置中，只有一个 ZooKeeper 和代理 id 实例。 配置步骤如下：（注意，以下过程中的topicName表示创建主题的名称，可以自己定义。）

（1）创建Kafka主题

bin/kafka-topicssh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topicName

创建主题后，会在 Kafka 代理终端窗口中获取通知，并在 config / serverproperties 文件中的“/ tmp / kafka-logs /"中指定创建主题的日志。

（2）启动生产者以发送消息

bin/kafka-console-producersh --broker-list localhost:9092 --topic topicName

生产者命令行客户端需要两个主要参数：

1代理列表（broker-list）： 要发送邮件的代理列表。 这种情况下，只有一个代理。

2监听端口： Config / serverproperties 文件包含代理端口 ID，可以查到代理正在侦听端口 9092，因此直接指定它。

生产者在 config / producerproperties 文件中指定默认生产者属性。

（3）启动消费者以接收消息

bin/kafka-console-consumersh --bootstrap-server localhost:9092 --topic topicName --from-beginning

消费者在config / consumerproperties 文件中指定了默认消费者属性。 打开一个新终端并键入以下消息消息语法。

（4）在生产者终端输入数据测试

生产者将等待消息的输入并发布到 Kafka 集群。 默认情况下，每行数据都作为新消息发布。在生产者终端输入数据，这些数据都会在消费者终端显示。

以上就是关于阿里大牛的Kafka动态配置了解下全部的内容，包括:阿里大牛的Kafka动态配置了解下、Kafka总学不明白：、Kafka的命令行工具等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！