kafka-03-主题的操作和动态配置变更

1 主题操作

使用 kafka-topics.sh 工具可以执行主题的大部分操作（配置变更部分已经被移动到 kafka-configs.sh 工具当中）。我们可以用 kafka-topics.sh 创建、修改、 删除和查看集群里的主题。要使用该工具的全部功能，需要通过 --bootstrap-server 参数提供 Kafka broker 的连接字符串 。

1.1 创建主题 --create

在集群里创建一个主题需要用到 3 个参数。这些参数是必须提供的，尽管有些已经有了 broker 级别的默认值。

1. 主题名字
   想要创建的主题的名字。

2. 复制系数
   主题的副本数量。

3. 分区
   主题的分区数量。

指定主题配置
可以在创建主题时显式地指定复制系数或者对配置进行覆盖 ，不过我们不打算在这里介绍如何做到这些。稍后会介绍如何进行配置覆盖 ， 它们是通过向 kafka-topics.sh 传递 --config 参数来实现的。本章还会介绍分区的重分配。

主题名字 可以包含字母、数字、下划线以及英文状态下的破折号和句号。

注意：主题名字的开头部分包含两个下划线是合法的，但不建议这么做。具有这种格式的主题一般是集群的内部主题（ 比如 __consumer_offsets 主题用于保存消费者群组的偏移量）。 也不建议在单个集群里使用英文状态下的句号和下划线来命名，因为主题的名字会被用在度量指标上，句号会被替换成下划线（比如“topic.1"会变成“topic_1"）。

试着运行下面的命令 ：
kafka-topics.sh --bootstrap-server <kafka connect> --create -- topic <string> --replication-factor <integer> --partitions <integer>

这个命令将会创建一个主题，主题的名字为指定的值 ， 并包含了指定数量的分区。集群会为每个分区创建指定数量的副本。
如果为集群指定了基于机架信息的副本分配策略， 那么分区的副本会分布在不同的机架上。如果不需要基于机架信息的分配策略，可以指定参数 --disable-rack-aware 。

示例 ：使用以下命令创建一个叫作 topic1 的主题 ， 主题包含 2 个分区，每个分区拥有1个副本。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic1 --replication-factor 1 --partitions 2
Created topic topic1.
[root@docker01 ~]# 

忽略重复创建主题的错误
在自动化系统里调用这个脚本时，可以使用 --if-not-exists 参数，这样即使主题已经存在，也不会抛出重复创建主题的错误。

1.2 列出主题详细信息 --describe

主题工具还能用来获取主题的详细信息。信息里包含了分区数量 、主题的覆盖配置以及每个分区的副本清单。 如果通过 --topic 参数指定特定的主题， 就可以只列出指定主题的详细信息。
示例 ： 列出集群里 topic1 主题的详细信息。

#查看主题 topic1 描述
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic1
Topic: topic1	PartitionCount: 2	ReplicationFactor: 1	Configs: 
	Topic: topic1	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
	Topic: topic1	Partition: 1	Leader: 0	Replicas: 0	Isr: 0

describe 命令还提供了一些参数 ，用于过滤输出结果 ， 这在诊断集群问题时会很有用 。不要为这些参数指定 --topic 参数 （ 因为这些参数的目的是为了找出集群里所有满足条件的主题和分区） 。 这些参数也无法与 list 命令一起使用（最后一部分会详细说明原因 ） 。
使用 --topics-with-overrides 参数可以找出所有包含覆盖配置的主题，它只会列出包含了与集群不一样配置的主题。
有两个参数可用于找出有问题的分区。使用 --under-replicated-partitions 参数可以列出所有包含不同步副本的分区 。 使用 --unavailable-partitions 参数可以列出所有没有首领的分区，这些分区已经处于离线状态，对于生产者和消费者来说是不可用的。

示例 ：列出包含不同步副本的分区。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions
[root@docker01 ~]# 

1.3 增加分区 --alter

有时候，我们需要为主题增加分区数量。主题基于分区进行伸缩和复制，增加分区主要是为了扩展主题容量或者降低单个分区的吞吐量。

调整基于键的主题
从消费者角度来看 ，为基于键的主题添加分区是很困难的。因为如果改变了分区的数量，键到分区之间的映射也会发生变化。所以，对于基于键的主题来说，建议在一开始就设置好分区数量，避免以后对其进行调整。

忽略主题不存在的错误
使用，--alter 命令修改主题时，如果指定了 --if-exists 参数，主题不存在的错误就会被忽略。如果要修改的主题不存在，该命令并不会返回任何错误。在主题不存在的时候本应该创建主题 ，但它却把错误隐藏起来，因此 不建议使用这个参数。

示例：将 topic1 主题的分区数量增加到 4 。

# 将 topic1 主题的分区数量增加到 4 	
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic topic1 --partitions 4
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
# 增加分区成功，再次查看主题 topic1 描述
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic1
Topic: topic1	PartitionCount: 4	ReplicationFactor: 1	Configs: 
	Topic: topic1	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
	Topic: topic1	Partition: 1	Leader: 0	Replicas: 0	Isr: 0
	Topic: topic1	Partition: 2	Leader: 0	Replicas: 0	Isr: 0
	Topic: topic1	Partition: 3	Leader: 0	Replicas: 0	Isr: 0
[root@docker01 ~]# 

减少分区数量
我们无法减少主题的分区数量。因为如果删除了分区，分区里的数据也一并被删除，导致数据不一致。 我们也无法将这些数据分配给其他分区，因为这样做很难，而且会出现消息乱序。所以，如果一定要减少分区数量，只能删除整个主题，然后重新创建它。

1.4 删除主题 --delete

如果一个主题不再被使用，只要它还存在于集群里，就会占用一定数量的磁盘空间和文件句柄。把它删除就可以释放被占用的资源。
为了能够删除主题， broker 的 delete.topic.enable 参数必须被设置为 true 。如果该参数被设为 false ，删除主题的请求会被忽略。
删除主题会丢弃主题里的所有数据。这是一个不可逆的操作，所以在执行时要十分小心。

示例 ：删除 topic1 主题 。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic topic1
Topic topic1 is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
[root@docker01 ~]# 

1.5 列出集群里的所有主题 --list

可以使用主题工具列出集群里的所有主题。每个主题占用一行输出， 主题之间没有特定的顺序 。

示例：列出集群里的所有主题。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
__consumer_offsets
# 新建主题 topic2
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic2 --replication-factor 1 --partitions 2
Created topic topic2.
# 再次查看所有主题
[root@docker01 /usr/local/kafka_2.13-2.6.0/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
__consumer_offsets
topic2
[root@docker01 ~]# 

2 动态配置变更

我们可以在集群处于运行状态时覆盖主题配置和客户端的配额参数。
我们打算在未来增加更多的动态配置参数，这也是为什么这些参数被单独放进了 kafka-configs.sh 。这样就可以为特定的主题和客户端指定配置参数。一旦设置完毕，它们就成为集群的永久配置，被保存在 Zookeeper 上， broker 在启动时会读取它们。不管是在工具里还是文档里，它们所说的动态配置参数都是基于“主题"实例或者 “客户端"实例的，都是可以被“覆盖"的。

2.1 覆盖主题的默认配置

为了满足不同的使用场景， 主题的很多参数都可以进行单独的设置。它们大部分都有 broker 级别的默认值 ，在没有被覆盖的情况下使用默认值。

更改主题配置的命令格式如下。

kafka-configs.sh --bootstrap-server <kafka connect> --alter --entity-type topics --entity-name <topic name> --add-config <key>=<value>[,<key>=<value>...] 

可用的主题配置参数（键）如下图所示。

示例 ： 将主题 topic1 的消息保留时间设为 1 个小时（ 3 600 000ms ）。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name topic2 --add-config retention.ms=3600000
Completed updating config for entity: topic 'topic2'.
[root@docker01 ~]# 

2.2 覆盖客户端的默认配置

对于 Kafka 客户端来说，只能覆盖生产者配额和消费者配额参数。这两个配额都以字节每秒为单位，表示客户揣在每个 broker 上的生产速率或消费速率。 也就是说，如果集群里有 5 个 broker ，生产者的配额是 10MB/s ，那么它可以以 10MB/s 的速率在单个 broker 上生成数据 ， 总共的速率可以达到 50MB/s 。

客户端 ID 与消费者群组
客户端 ID 可以与消费者群组的名字不一样。消费者可以有自己的 ID ，因此不同群组里的消费者可能具有相同的 ID 。在为消费者客户端设置 ID 时，最好使用能够表明它们所属群组的标识符，这样便于群组共享配额，从日志里查找负责请求的群组也更容易一些。

更改客户端配置的命令格式如下：

kafka-configs.sh --bootstrap-server <kafka connect> --alter --entity-type clients --entity-name <client ID> --add-config <key>=<value>[,<key>=<value> ...] 

可用的客户端配置参数（键）：

配置项	描述
producer_bytes_rate	单个生产者每秒钟可以往单个 broker 上生成的消息字节数
consumer_bytes_rate	单个消费者每秒钟可以从单个 broker 读取的消息字节数

2.3 列出被覆盖的配置

使用命令行工具可以列出所有被覆盖的配置，从而用于检查主题或客户端的配置。与其他工具类似，这个功能通过 --describe 命令来实现。

示例 ： 列出主题 topic2 所有被覆盖的配置。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type topics --entity-name topic2
Dynamic configs for topic topic2 are:
  retention.ms=3600000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=3600000}
[root@docker01 ~]# 

只能显示主题的覆盖配置
这个命令只能用于显示被覆盖的配置，不包含集群的默认配置。目前还无法通过 Zookeeper 或 Kafka 实现动态地获取 broker 本身的配置。

2.4 移除被覆盖的配置

动态的配置完全可以被移除，从而恢复到集群的默认配置。可以使用 --alter 命令和 --delete-config 参数来删除被覆盖的配置。

示例 ：删除主题 topic2 的 retention.ms 覆盖配置。

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type topics --entity-name topic2
Dynamic configs for topic topic2 are:
  retention.ms=3600000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=3600000}
# 删除主题 topic2 的 retention.ms 覆盖配置
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name topic2 --delete-config retention.ms
Completed updating config for topic topic2.
[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type topics --entity-name topic2
Dynamic configs for topic topic2 are:
[root@docker01 ~]# 

3 消费者群组

Kafka 里，有两个地方保存着消费者群组的信息。对于旧版本的消费者来说，它们的信息保存在 Zookeeper 上;对于新版本的消费者来说，它们的信息保存在 broker 上。 kafka­-consumer-groups.sh 工具可以用于列出上述两种消费者群组。 它也可以用于删除消费者群组和偏移量信息，不过这个功能仅限于旧版本的消费者群组（信息保存在 Zookeeper 上） 。
在对旧版本的消费者群组进行操作时，需要通过 --zookeeper 参数指定 Zookeeper 的地址;
在对新版本的消费者群组进行操作时，则需要使用 --bootstrap-server 参数指定 broker 的主机名和端口。

3.1 列出并描述群组

使用 --bootstrap-server 、--list 参数。

示例 ： 列出kafka集群的消费者群组的信息，下面显示有3个消费者群组

[root@docker01 ~]# /usr/local/kafka_2.13-2.6.0/bin/kafka-consumer-groups.sh  --bootstrap-server localhost:9092 --list
test-group
test-group2
test-group1
[root@docker01 ~]#