更多内容，前往 IT-BLOG
主题操作
使用 kafka-topics.sh 工具可以执行主题的大部分操作。可以用它创建、修改、删除和查看集群里的主题。要使用该工具的全部功能，需要通过 --zookeeper参数提供 Zookeeper的连接字符串。
kafka 的大部分命令行工具直接操作 Zookeeper 上的元数据，并不会连接到 broker上。因此，要确保所使用工具版本与集群里的 broker版本相匹配。直接使用集群 broker自带的工具是最保险的。
创建主题
在集群中创建一个主题需要3个参数：主题名字（可以包含字母、数字、下划线以及英文状态下的破折号和句号），复制系数（主题的副本数量），分区（主题的分区数量）
主题命名的开头部分包含两个下划线是合法的，但不建议这么做。具有这种格式的主题一般是集群的内部主题（比如__consumer_offsets 主题用于保存消费者群组的偏移量）。也不建议在单个集群里使用英文状态下的句号和下划线来命名，因为主题的名字会被在度量指标上，句号会被替换成下划线（比如“topic.1会变成topic_1”）
如下命令会创建一个主题，主题的名字为指定的值，并包含了指定数量的分区。集群会为这个分区创建指定数量的副本。如果为集群指定了基于机架信息的分配策略，可以指定参数 --disable-rack-aware
kafka-topic.sh --zookeeper <zookeeper connect> --create --topic <string> --replication-factor <integer> --partitions <integer>
示例：使用以下命令创建一个叫做 my-topic 主题，主要包括八个分区，每个分区拥有两个副本。
kafka-topic.sh --zookeeper hadoop1:2181 --create --topic my-topic --replication-factor 2 --partitions 8
忽略重复创建主题的错误：在自动化系统里调用这个脚本时，可以使用 --if-not-exists参数，这样即使主题已经存在，也不会抛出重复创建主题的错误。
增加分区
有时我们需要为主题增加分区数量，主题基于分区进行伸缩和复制，增加分区主要是为了扩展主题容量和降低单个分区的吞吐量。如果要在单个消费者组内运行更多的消费者，那么分区数量也需要响应增加，因为一个分区只能由群组里的一个消费者读取。
调整基于键的主题：从消费者角度来看，为基于键的主题添加分区是很困难。因为如果改变了分区的数量，键到分区之间的映射也会发生改变。所以，基于键的主题来说，建议一开始就设置好分区数量，避免以后对其进行调整。
示例：将 my-topic 主题的分区增加到16个。
kafka-topic.sh --zookeeper hadoop1:2181 --alter --topic my-topic --partitions 16
减少分区数量：我们无法减少主题的分区数量。因为如果删除了分区，分区里的数据也一并被删除，导致数据不一致。我们也无法将这些数据分配给其他分区，因为这样做很难，而且会出现消息乱序。所以，如果一定要减少分区数量，只能删除整个topic，然后重新创建它。
删除主题
如果一个主题不再被使用，只要它还在集群中，就会占用一定数量的磁盘空间和文件句柄。把它删除就可以释放占用的资源。为了能够删除主题，broker 的 delete.topic.enable 参数必须设置为 true。如果为 false，删除主题的请求会被忽略。删除主题会丢弃主题里的所有数据。这是一个不可逆的操作，执行时需要十分消息。
示例：删除 my-topic 主题。
kafka-topic.sh --zookeeper hadoop1:2181 --delete --topic my-topic
列出集群里的所有主题
可以使用主题工具列出集群里的所有主题。每个主题占用一行输出，主题之间没有特定的顺序。
示例：列出集群里的所有主题
kafka-topic.sh --zookeeper hadoop1:2181 --list
主题工具还能用来获取主题的详细信息。信息里包含了分区数量、主题的覆盖配置以及每个分区的副本清单。如果通过 --topic参数指定特定的主题，就可以只列出指定主题的详细信息。示例：列出集群里所有主题的详细信息
kafka-topic.sh --zookeeper hadoop1:2181 --describe
describe 命令还提供了一些参数，用于过滤输出结果，这个在诊断集群问题时会很有用。不要为这些参数指定 --topic参数。这些参数也无法与 list命令一起使用。使用 --topics-with-overrides 参数可以找出所有包含覆盖配置的主题，它只会列出包含了与集群不一样配置的主题。
有两个参数可用于找出问题的分区。使用 --under-replicated-partitions 参数可以列出所有包含不同步副本的分区。使用 --unavaliable-partitions 参数可以列出所有没有首领的分区，这些分区已经处理离线状态，对于生产者和消费者来说是不可用的。
示例：列出包含不同步副本的分区
kafka-topic.sh --zookeeper hadoop1:2181 --describe --under-replicated-partitions
消费者群组
kafka中有两个地方保存着消费者群组的消息，旧版本的消费者来说，它的信息保存在 zk上。对于新版本的消费者来说，它的信息保存在 broker上。kafka-consumer-group.sh 工具可以用于列出上述两种消费者群组。也可以用于删除消费者群组和偏移量信息，不过这个功能仅限于旧版本的消费者群组（信息保存在zk上）。在对旧版本的消费者群组进行操作时，需要通过 --zookeeper 参数指定 Zookeeper的地址；在对新版本的消费者群组进行操作时，则需要使用 --bootstrap-server参数指定 broker的主机名和端口。
列出并描述群组
旧版本的消费者群组，使用 --zookeeper 和 --list参数列出消费者群组；示例：列出旧版本的消费者群组
kafka-consumer-group.sh --zookeeper hadoop1:2181 --list
新版本的消费者群组，使用 --bootstrap-server、--list 和 --new-consumer 参数。
kafka-consumer-group.sh --new-consumer --bootstrap-server hadoop1:9092 --list
对于任意群组来说，使用 --describe 代替 --list，并通过 --group 指定特定的群组，就可以获取该群组的详细信息。它会列出群组里所有主题的信息和分区的偏移量。示例：
kafka-consumer-group.sh --new-consumer --bootstrap-server hadoop1:9092 --describe --group myGroup
输出字段解释：
| 字段 | 描述 |
| GROUP | 消费者群组的名字 |
| TOPIC | 正在读取的主题的名字 |
| PARTITION | 正在读取的分区 ID |
| CURRENT-OFFSET | 消费者群组最近提交的偏移量，也就是消费者在分区里读取的当前位置 |
| LOG-END-OFFSET | 当前高水位偏移量，也就是最近一个被读取消息的偏移量，同时也是最近一个被提交到集群的偏移量 |
| LAG | 消费者的 CURRENT-OFFSET 和 broker 的 LOG-END-OFFSET之间的差距 |
| OWNER | 消费者群组里正在读取该分区的消费者。这是一个消费者的ID，不一定包含消费者的主机名 |
删除群组
只有旧版本支持删除群组操作。删除群组将在 zk上移除整个群组，包括所有已保存的偏移量。在执行该操作之前，必须关闭所有的消费者。如果不关闭，可能会导致消费者出现不可预期的行为，因为群组的元数据已经从 Zookeeper 上移除了。示例：删除消费者群组 testgroup
kafka-consumer-group.sh --zookeeper hadoop1:2181 --delete --group testgroup
该命令也可以用于删除单个主题的偏移量。再次强调，在进行删除操作之前，需要关闭消费者，或者不要让他们读取即将被删除的主题。示例：从消费者群组 testgroup 里删除 my-topic 主题的偏移量。
kafka-consumer-group.sh --zookeeper hadoop1:2181 --delete --group testgroup --topic my-topic
偏移量管理
可以获取偏移量，并保存批次的最新偏移量，从而实现偏移量的重置。在需要重新读取消息或者因消费者无法正常处理消息(比如包含了非法格式的消息)需要跳过偏移量时，需要进行偏移量重置。
管理已提交到 kafka 的偏移量：目前，还没有工具可以用于管理由消费者客户端提交到 Kafka 的偏移量，管理功能只对提交到 zk的偏移量可用。另外，为了能够管理提交到 Kafka的消费者群组偏移量，需要在客户端使用响应的 API来提交群组的偏移量。
导出偏移量：kafka 没有为导出偏移量提供线程的脚本，不过可以使用 kafka-run-class.sh 脚本调用底层的 Java类来实现导出。在导出偏移量时，会生成一个文件，文件里包含了分区和偏移量的信息。偏移量信息以一种导入工具能够识别的格式保存在文件里。每个分区在文件里占用一行，格式为：/consumer/GROUPNAME/offsets/topic/TOPICNAME/PARTITIONID-0:OFFSET。示例：将群组 testgroup 的偏移量导出到 offsets 文件里。
1 # kafka-run-class.sh kafka.tools.ExportZkOffsets --zkconnect hadoop1:2181 --group testgroup --output-file offsets
2 # cat offsets
3 /consumers/testgroup/offsets/my-topic/0:8904
导入偏移量：使用导出的文件重置消费者群组的偏移量。一般情况下，我们会导出消费者群组的当前偏移量，并将导出的文件复制一份(备份)，然后修改复制文件里的偏移量。这里要注意，在使用导入命令时，不需要使用 --group 参数，因为文件里已经包含了消费者群组的名字。需要注意，要先关闭消费者，如果消费者群组处于活跃状态，它们不会读取新的偏移量，反而有可能将导入的偏移量覆盖掉。从 offsets 文件里将偏移量导入到消费者群组 testgroup
kafka-run-class.sh kafka.tools.ImportZkOffsets --zkconnect hadoop1:2181 --input-file offsets
动态配置变更
我们可以在集群处于运行状态时覆盖主题配置和客户端的配置参数。为了会增加更多的配置参数，这也是为什么这些参数被单独放进 kafka-configs.sh。这样就可以为特定的主题和客户端指定配置参数。一旦设置完毕，它就成为集群的永久配置，被保存在 Zookeeper 上，broker 在启动时会读取它们。不管是在工具里还是文档里，他们所说的动态配合参数都是基于“主题”实例或者“客户端”实例的，都是可以被“覆盖”的。这里也需要提供 --zookeeper 参数提供 Zookeeper集群的连接字符串。
覆盖主题的默认配置
为了满足不同的使用场景，主题的很多参数都可以进行单独的设置。它们大部分都有 broker 级别的默认值，在没有被覆盖的情况下使用默认值。更改主题配置的命令格式如下：kafka-config.sh
kafka-configs.sh --zookeeper hadoop1:2181 --alter --entity-type topics --entity-name <topic name> --add-config <key>=<value>[,<key>=<value>...]
| 配置项 | 描述 |
| cleanup.policy | 如果被设置为 compact，只有最新包含了指定 key的消息会被保留下来（压缩日志），其他的被丢弃掉。 |
| compression.type | broker 在将消息批次写入磁盘时所使用的压缩类型，目前支持 gzip、lz4、snappy |
| delete.retention.ms | 被标识为待删除的数据能够保留多久，以 ms为单位，该参数只对压缩日志类型的主题有效 |
| file.delete.delay.ms | 从磁盘上删除日志片段和索引之前可以等待多长时间，以 ms为单位 |
| flush.message | 需要收到多少个消息才能将他们刷新到磁盘 |
| flush.ms | 在将消息刷新到磁盘之前可以等待多长时间，以 ms为单位 |
| index.interval.bytes | 日志片段的两个索引之间能够容纳的消息字节数 |
| max.message.bytes | 最大的消息字节数 |
| message.format.version | broker 将消息写入磁盘时所使用的消息格式，必须是有效的 API版本号（比如 0.10.0） |
| message.timestamp.difference.max.ms | 消息自带的时间戳和 broker收到消息时的时间戳之间最大的差值，以 ms为单位。该参数只在 message.timestamp.type 被设为 Create-Time 时有效 |
| message.timestamp.type | 在将消息写入磁盘时使用哪一种时间戳。目前支持两种值，其中 CreateTime指客户端指定的时间戳，而LogAppendTime 指消息被写入分区时的时间戳 |
| min.cleanable.dirty.ratio | 日志压缩器压缩分区的频率，使用末压缩日志片段数与总日志片段数之间的比例来表示。该参数只对压缩日志类型的主题有效可用分区的最少同步副本 |
| preallocate | 如果被设为 true，需要为新的日志片段预分配空间 |
| retention.bytes | 主题能够保留的消息量，以字节为单位 |
| retention.ms | 主题需要保留消息多长时间，以 ms为单位 |
| segment.bytes | 日志片段的消息字节数 |
| segment.ms | 日志片段多长时间滚动一次，以 ms为单位 |
| segment.index.bytes | 每个日志片段的最大索引字节数 |
| segment.jitter.ms | 滚动日志片段时，在 segment.ms 基础上增加了随机毫秒数 |
| unclean.leader.election.enable | 如果被设为 true，不彻底的首领选择无效 |
| min.insync.replicas | 可用分区的最少同步副本 |
示例：将主题 my-topic 的消息保留时间设为1个小时
kafka-configs.sh --zookeeper hadoop1:2181 --alter --entity-type topics --entity-name my-topic --add-config retention.ms=3600000
覆盖客户端的默认配置
对于 kafka客户端而言，只能覆盖生产者配额和消费者配额参数。这两个配额都以字节每秒为单位，表示客户端在每个 broker上的生产速率或消费速率。也就是说，如果集群里有 5个broker，生产者的配额是 10MB/s 的速率在单个 broker上生产数据，总共的速率可以达到 50MB/s。
kafka-configs.sh --zookeeper hadoop1:2181 -alter --entity-type clients --entity-name <client ID> --add-config <key> = <value>[,<key> = <value>...]
可用的客户端配置参数(键)如下所示：
| 配置项 | 描述 |
| producer_bytes_rate | 单个生产者每秒钟可以往单个 broker上生产的消息字节数 |
| consumer_bytes_rate | 单个消费者每秒钟可以从单个 broker 读取的消息字节数 |
列出被覆盖的配置
使用命令工具可以列出所有被覆盖的配置，从而用于检查主题或客户端的配置。与其他工具类似，这个功能通过 --describe 命令来实现。示例：列出主题 my-topic 所有被覆盖的配置
kafka-configs.sh --zookeeper hadoop1:2181 --describe --entity-type topics --entity-name my-topic
移除被覆盖的配置
动态的配置完成可以被移除，从而恢复到集群的默认配置。可以使用 --alter 命令和 --delete-config 参数来删除被覆盖的配置。示例：删除主题 my-topic 的 retention.ms 覆盖配置
kafka-configs.sh --zookeeper hadoop1:2181 --alter --entity-type topics --entity-name my-topic --delete-config retention.ms
分区管理
kafka提供了两个脚本管理用于管理分区，一个用于重新选举首领，另一个用于将分区分配给 broker。从而实现集群流量的负载均衡。
首选的首领选举
使用多个分区副本可以提升可靠性。不过，只有其中一个副本可以成为分区首领，而且只有首领所在的 broker 可以进行生产和消费活动。kafka 将副本清单里面的第一个同步副本选为首领，但是在关闭重启 broker之后，并不会自动恢复原先首领的身份。
触发首选的副本选举，可以让 broker重新获取首领。当该事件被触发时，集群控制器会为分区重新选择理想的首领。选举过程不会造成负面影响，因为客户端可以自动跟踪首领的变化。也可以通过工具手动触发选举。
kafka-preferred-replica-election.sh --zookeeper hadoop1:2181
修改分区副本
有些时候，可能需要修改分区的副本，一下是需要修改分区副本的场景：
【1】主题分区在整个集群里的不均衡分区造成了集群负载的不均衡；
【2】broker 离线造成分区不同步；
【3】新加入的 broker需要从集群里获得负载；
可以使用 kafka-reassign-partitions.sh工具来修改分区。使用该工具需要经过如下步骤：第一步，根据 broker清单和主题清单生成一组迁移步骤；第二步，执行这些迁移步骤。第三步可选，可以使用生成的迁移步骤验证分区重分配的进度和完成情况。为了生成迁移步骤，需要先创建一个包含了主题清单的 JSON文件，文件格式如下：
1 {
2 "topics": [
3 {
4 "topic": "foo"
5 },
6 {
7 "topic": "foo1"
8 }
9 ],
10 "version": 1
11 }
示例：为 topic.json 文件里的主题生成迁移步骤，以便将这些主题迁移到 broker0 和 broker1上。broker 的ID以逗号分隔，并作为参数提供给命令工具。这个工具会在标准控制台上输出两个 JSON对象，分别描述了当前的分区分配情况以及建议的分区分配方案。这些 JSON对象的格式如下：{"partitions": [{"topic": "my-topic","partition": 0,"replicas": [1,2]}], "version":1}可以把第一个 JSON 对象保存起来，以便在必要的时候进行回滚。第二个 JSON对象应该被保存到另一个文件里，作为 kafka-reassign-partitions.sh 工具的输入来执行第二个步骤。
kafka-reassign-partitions.sh --zookeeper hadoop1:2181 --generate --topics-to-move-json-file topic.json --broker-list 0,1
示例：使用 reassign.json 来执行建议的分区分配方案。该命令会将指定分区的副本重新分配到新的 broker上。集群控制器通过为每个分区添加新副本实现重新分配（增加复制系数）。新的副本将从首领哪里复制所有数据。根据分区大小不同，复制过程可能需要花一些时间，因为数据是通过网络复制到新副本上的。在复制完成之后，控制器将旧副本从副本清单里面移除。
kafka-reassign-partitions.sh --zookeeper hadoop1:2181 --execute --reassignment-json-file reassign.json
如果要从单个broker上移除多个分区，比如将 broker移除集群，那么在重新分配副本之前最好先关闭或者重启 broker。这样，这个 broker就不再是任何一个分区的首领，它的分区就可以被分配到集群里的其他 broker上。
重分配完成之后，可以通过 kafka-reassign-partitions.sh 工具验证重分配的状态。它可以显示重分配的进度、已完成重分配的分区以及错误信息，为了实现这一点，需要在执行过程中使用 JSON对象文件。示例：验证 reassign.json 文件里指定的分区重分配情况。
kafka-reassign-partitions.sh --zookeeper hadoop1:2181 --verify --reassignment-json-file reassign.json
分区重分配对集群的性能影响很大，因为它会引起内存页缓存发生变化，并占用额外的网络和磁盘资源。将重分配过程拆分成多个小步骤可以将这种影响降到最低。
修改复制系数
分区重新分配工具提供了改变分区复制系数的特性。如果在创建分区时指定了错误的复制系数（比如创建主题时没有足够多的broker）那么就有必要修改它们。可以通过创建 JSON对象来完成，该对象使用分区重分配的执行步骤中使用的格式，显示指定分区需要的副本数量。集群将完成重分配过程，并使用新的复制系数。例如，假设主题 my-topic 有一个分区，该分区的复制系数为1。
1 {
2 "partitions": [
3 {
4 "topics": "my-topic",
5 "partition": 0,
6 "replicas": [1]
7 }
8 ],
9 "version": 1
10 }
在分区重新分配的执行步骤中使用以下 JSON可以将复制系数改为2。也可以通过类似的方法减少分区的复制系数。
1 {
2 "partitions": [
3 {
4 "partition": 0,
5 "replicas": [1,2],
6 "topic": "my-topic"
7 }
8 ],
9 "version": 1
10 }
