Kafka 命令行操作

Kafka Topic、生产者、消费者、消费者组、分区和副本调整命令实践。

Kafka 命令行操作

2.1 主题命令行操作

1）查看操作主题命令参数

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh

参数	描述
--bootstrap-server <String: server toconnect to>	连接的Kafka Broker主机名称和端口号。
--topic <String: topic>	操作的topic名称。
--create	创建主题。
--delete	删除主题。
--alter	修改主题。
--list	查看所有主题。
--describe	查看主题详细描述。
--partitions <Integer: # of partitions>	设置分区数。
--replication-factor<Integer: replication factor>	设置分区副本。
--config <String: name=value>	更新系统默认的配置。

2）查看当前服务器中的所有topic

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --list

3）创建first topic

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --create --partitions 6 --replication-factor 2 --topic first

选项说明：

--topic 定义topic名

--replication-factor 定义副本数，副本数量，一定不能超过 broker的数量

--partitions 定义分区数，分区的数量，在生产环境，是broker 的1.5 倍 ~ 2倍

4）查看first主题的详情

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --describe --topic first

字段	具体含义	关键解读
`PartitionCount: 6`	该 Topic 有 6 个分区	分区数越多，并行读写能力越强（6 个分区可同时处理 6 路读写请求）
`ReplicationFactor: 2`	每个分区有 2 个副本（1 个 Leader + 1 个 Follower）	副本因子 = 2 意味着每个分区有 1 个备份，某台 Broker 宕机后，副本可切换 Leader，保证服务不中断
`Configs: segment.bytes=1073741824`	日志段文件大小为 1GB（默认值）	达到 1GB 会生成新的日志文件，方便 Kafka 清理 / 压缩旧数据

字段	具体含义
`Partition: 0`	分区编号（0-5 共 6 个分区）
`Leader: 2`	该分区的主副本（Leader）在 Broker 2 上（所有读写请求都走 Leader）
`Replicas: 2,1`	该分区的所有副本列表：Broker 2（Leader）、Broker 1（Follower）
`Isr: 2,1`	同步副本列表（In-Sync Replica）：Broker 2 和 1 都与 Leader 数据同步，属于可用副本

5）修改分区数（注意：分区数只能增加，不能减少）

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --alter --topic first --partitions 6

6）再次查看first主题的详情

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --describe --topic first

7）删除topic（学生自己演示）

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --bootstrap-server kafka102:9092 --delete --topic first

2.2 生产者命令行操作

1）查看操作生产者命令参数

[root@kafka102 kafka-3.2.1]$ bin/kafka-console-producer.sh

参数	描述
--bootstrap-server <String: server toconnect to>	连接的Kafka Broker主机名称和端口号。
--topic <String: topic>	操作的topic名称。

2）发送消息

[root@kafka102 kafka-3.2.1]$ bin/kafka-console-producer.sh --bootstrap-server kafka102:9092 --topic first
>hello world
>kafka kafka

2.3 消费者命令行操作

1）查看操作消费者命令参数

[root@kafka102 kafka-3.2.1]$ bin/kafka-console-consumer.sh

参数	描述
--bootstrap-server <String: server toconnect to>	连接的Kafka Broker主机名称和端口号。
--topic <String: topic>	操作的topic名称。
--from-beginning	从头开始消费。
--group <String: consumer group id>	指定消费者组。

2）消费first主题中的数据

[root@kafka102 kafka-3.2.1]$ bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092 --topic first

3）把主题中所有的数据都读取出来（包括历史数据）

[root@kafka102 kafka-3.2.1]$ bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092 --from-beginning --topic first

2.4 消费者组

1.作用

Kafka 消费者组= 一组一起消费消息的消费者，共用一个组名。

同一个组：分工合作、负载均衡

不同组：各玩各的、互不干扰

Topic（主题）
├── Partition 0
├── Partition 1
└── Partition 2

Consumer Group（消费组）
├─ 消费者A  ← 消费 分区0、1
└─ 消费者B  ← 消费 分区2

2.实操

创建分组

没有分组,自动创建分组

[root@kafka103 kafka-3.2.1]#  bin/kafka-console-consumer.sh --bootstrap-server kafka103:9092 --group mygroup  --topic first

[root@kafka104 kafka-3.2.1]#  bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092 --group my_new_consumer_group  --topic first

注意：暂无补充。

查看分组列表

[root@kafka103 bin]# ./kafka-consumer-groups.sh --bootstrap-server kafka102:9092 --list

查看分组详情

[root@kafka103 bin]# ./kafka-consumer-groups.sh --bootstrap-server kafka102:9092 --describe --group mygroup

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
mygroup         first           0          3               3               0               console-consumer-10076212-677c-4fc2-8d61-82dea3861eba /192.168.2.103  console-consumer
mygroup         first           1          6               6               0               console-consumer-10076212-677c-4fc2-8d61-82dea3861eba /192.168.2.103  console-consumer
mygroup         first           2          8               8               0               console-consumer-10076212-677c-4fc2-8d61-82dea3861eba /192.168.2.103  console-consumer

再次添加一个节点

[root@kafka104 kafka-3.2.1]# bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092 --topic first --group mygroup

查看详情

[root@kafka103 bin]# ./kafka-consumer-groups.sh --bootstrap-server kafka102:9092 --describe --group mygroup

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
mygroup         first           0          3               3               0               console-consumer-10076212-677c-4fc2-8d61-82dea3861eba /192.168.2.103  console-consumer
mygroup         first           1          6               6               0               console-consumer-10076212-677c-4fc2-8d61-82dea3861eba /192.168.2.103  console-consumer
mygroup         first           2          8               8               0               console-consumer-bef537e8-15d4-497b-a33e-068786262515 /192.168.2.104  console-consumer

解释 mygroup 消费状态全正常，2 个消费者分工干活，没有消息堆积！

字段 / 值	学生能懂的意思
GROUP: mygroup	这个消费小队的名字叫 mygroup
TOPIC: first	小队在处理名叫 first 的消息仓库
PARTITION: 0/1/2	仓库分 3 个货架（0、1、2 号）
CURRENT-OFFSET: 3/6/8	各货架已消费到第 3/6/8 条消息
LOG-END-OFFSET: 3/6/8	各货架最新一条消息是第 3/6/8 条
LAG: 0	✅ 红重点：消费没掉队，0 条消息堆积
CONSUMER-ID 有 2 个	小队里有 2 个消费者（2 个干活的人）
HOST: 103/104	1 个消费者在 103 机器、1 个在 104 机器
分区 0/1 → 103 机器	103 机器的消费者负责 0、1 号货架
分区 2 → 104 机器	104 机器的消费者负责 2 号货架

注意:
  如果指定了kafka102 但是host可能是192.168.2.103/104
--bootstrap-server Kafka 集群的连接入口（消费者用来获取集群元数据的 Broker 地址）
HOST 是 消费者实例实际运行的服务器 IP（执行消费命令的机器）

2.5 案例实操（脚本版本）

1）启动kafka服务

[root@kafka102 zookeeper-3.5.9]$ zk.sh start
[root@kafka102 kafka-3.2.1]$ xkafka.sh start
---启动 root@kafka102 kafka---
---启动 root@kafka103 kafka---
---启动 root@kafka104 kafka---
[root@kafka102 ~]# xjps.sh
======root@kafka102======
2116 QuorumPeerMain
4956 Kafka
5855 Jps
======root@kafka103======
5206 Kafka
2220 QuorumPeerMain
5918 Jps
======root@kafka104======
5251 Kafka
13299 Jps
2280 QuorumPeerMain

2）查看kafka版本

[root@kafka102 kafka-3.2.1]$ bin/kafka-topics.sh --version3.2.1 (Commit:b172a0a94f4ebb9f)

3）创建测试topic

[root@kafka102 kafka-3.2.1]$ xkafka-topic-create.sh test01
Created topic test01.
[root@kafka102 kafka-3.2.1]$ xkafka-topic-create.sh test02
Created topic test02.

4）查看topic

[root@kafka102 kafka-3.2.1]$ xkafka-topic-list.sh
test01
test02

5）查看topic详情

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test02
Topic: test02	TopicId: RxKefgtYQq6bZ7Bo3Kx7Zg	PartitionCount: 6	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test02	Partition: 0	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test02	Partition: 1	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test02	Partition: 2	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test02	Partition: 3	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: test02	Partition: 4	Leader: 0	Replicas: 0,2	Isr: 0,2
	Topic: test02	Partition: 5	Leader: 2	Replicas: 2,1	Isr: 2,1

6）删除 Topic

[root@kafka102 kafka-3.2.1]$ xkafka-topic-delete.sh test02
[2022-08-23 16:12:30,738] WARN [AdminClient clientId=adminclient-1] Connection to node -3 (kafka104/192.168.2.104:909) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[root@kafka102 kafka-3.2.1]$ xkafka-topic-list.sh
test01

7）生产topic

[root@kafka102 kafka-3.2.1]$ xkafka-producer.sh test03
>I love kafka

8）消费topic

[root@kafka102 kafka-3.2.1]$ xkafka-consume.sh test03
I love kafka

9）查看自动创建topic详情

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test03
Topic: test03	TopicId: FUhVq2otTQqC103_PxszUQ	PartitionCount: 1	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: test03	Partition: 0	Leader: 2	Replicas: 2	Isr: 2

10）修改topic分区

注意：分区数只能增加，不能减少

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test
Topic: test	TopicId: WZAkLUuRRJqUn-wbJH5iRA	PartitionCount: 6	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: test	Partition: 1	Leader: 0	Replicas: 0,2	Isr: 0,2
	Topic: test	Partition: 2	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: test	Partition: 3	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test	Partition: 4	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test	Partition: 5	Leader: 2	Replicas: 2,0	Isr: 2,0

[root@kafka102 kafka-3.2.1]$ xkafka-topic-update-partition.sh test 8

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test
Topic: test	TopicId: WZAkLUuRRJqUn-wbJH5iRA	PartitionCount: 8	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: test	Partition: 1	Leader: 0	Replicas: 0,2	Isr: 0,2
	Topic: test	Partition: 2	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: test	Partition: 3	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test	Partition: 4	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test	Partition: 5	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test	Partition: 6	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test	Partition: 7	Leader: 2	Replicas: 2,0	Isr: 2,0

2.6 修改 Topic 副本数

分区修改只能改大，而副本可以改大也可以改小

1）创建increase-replication-factor.json文件

[root@kafka102 kafka-3.2.1]$ touch increase-replication-factor.json

2）编辑increase-replication-factor.json文件

[root@kafka102 kafka-3.2.1]$ vim increase-replication-factor.json
{
    "version":1,
    "partitions":[
        {"topic":"test","partition":0,"replicas":[0,1,2]},
        {"topic":"test","partition":1,"replicas":[0,1,2]},
        {"topic":"test","partition":2,"replicas":[0,1,2]},
        {"topic":"test","partition":3,"replicas":[0,1,2]},
        {"topic":"test","partition":4,"replicas":[0,1,2]},
        {"topic":"test","partition":5,"replicas":[0,1,2]}
    ]
}

3）创建新的test topic

[root@kafka102 kafka-3.2.1]$ xkafka-topic-delete.sh test
[root@kafka102 kafka-3.2.1]$ xkafka-topic-list.sh
[root@kafka102 kafka-3.2.1]$ xkafka-topic-create.sh test
[root@kafka102 kafka-3.2.1]$ xkafka-topic-list.sh

4）查看topic详情

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test
Topic: test	TopicId: z2Un9ZjtQUmyRcnbs4HLVA	PartitionCount: 6	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 0	Replicas: 0,2	Isr: 0,2
	Topic: test	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: test	Partition: 2	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: test	Partition: 3	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test	Partition: 4	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test	Partition: 5	Leader: 1	Replicas: 1,2	Isr: 1,2

5）修改副本

[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --reassignment-json-file increase-replication-factor.json --execute

6）查看topic详情

[root@kafka102 kafka-3.2.1]# xkafka-topic-describe.sh test
Topic: test     TopicId: 5RHAIebLS56CsOqhv0kVDg PartitionCount: 6       ReplicationFactor: 3    Configs: segment.bytes=1073741824
        Topic: test     Partition: 0    Leader: 1       Replicas: 0,1,2 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 0       Replicas: 0,1,2 Isr: 0,2,1
        Topic: test     Partition: 2    Leader: 2       Replicas: 0,1,2 Isr: 2,1,0
        Topic: test     Partition: 3    Leader: 1       Replicas: 0,1,2 Isr: 1,2,0

7）修改json文件

[root@kafka102 kafka-3.2.1]$ vim increase-replication-factor.json
{
    "version":1,
    "partitions":[
        {"topic":"test","partition":0,"replicas":[1,2]},
        {"topic":"test","partition":1,"replicas":[0,2]},
        {"topic":"test","partition":2,"replicas":[0,1]},
        {"topic":"test","partition":3,"replicas":[1,2]},
        {"topic":"test","partition":4,"replicas":[0,2]},
        {"topic":"test","partition":5,"replicas":[0,1]}
    ]
}

8）修改副本

[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --reassignment-json-file increase-replication-factor.json --execute

9）查看topic详情

[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh test
Topic: test	TopicId: z2Un9ZjtQUmyRcnbs4HLVA	PartitionCount: 6	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 1	Replicas: 1,2	Isr: 2,1
	Topic: test	Partition: 1	Leader: 0	Replicas: 0,2	Isr: 2,0
	Topic: test	Partition: 2	Leader: 0	Replicas: 0,1	Isr: 1,0
	Topic: test	Partition: 3	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test	Partition: 4	Leader: 0	Replicas: 0,2	Isr: 2,0
	Topic: test	Partition: 5	Leader: 0	Replicas: 0,1	Isr: 1,0