经常用到的 Kafka 命令记录以下，方便查找。kafka 的搭建可以参考这篇文章 Kafka和zookeeper搭建

注意: –bootstrap-server 参数配置的 kafka 地址需要根据 kafka 监听地址来判断, 如果监听的是自己的 ip 而不是 0.0.0.0, 那就不要配置 localhost, 改用 ip 地址.

Topic 操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 创建topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 3 --topic test_topic


# 查看topic具体信息
bin/kafka-topics.sh --bootstrap-server localhost:9092  --describe --topic test_topic


# 查看所有的topic列表
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list


# 删除topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test_topic


# 修改topic分区数
bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --partitions 8 --topic test_topic

kafka 删除 topic 有可能提示 marked for deletion 这种情况下, 需要停止所有的生产者和消费者

消费组操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 显示所有消费组
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --all-groups --list
    

# 查看所有消费组详情
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --all-groups --describe
    

# 查看单独组的详细信息
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --describe
    

# 手动创建消费组
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic idkNfcKey1 --consumer-property group.id=idpTsp1
    

# 删除消费组（前提是没有在消费的）
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group  --delete

控制台生产消费

1
2
3
4
5
6
# 生产者发送消息, 然后控制台输入内容，回车即可
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test_topic


# 消费者消费消息, --from-beginning  是指从头进行消费，可能会打印大量东西在控制台上。
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic --from-beginning

控制台查看最新消息

查看一个 topic 的每个 partition 的 offset

1
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 1.1.1.1:9092,2.2.2.2:9092,3.3.3.3:9092 --topic nova_event_topic --time -1

image.png

根据每个 partition 的 offset 来消费最新消息

1
bin/kafka-console-consumer.sh --bootstrap-server 1.1.1.1:9092,2.2.2.2:9092,3.3.3.3:9092 --topic nova_event_topic --partition 0 --offset 12080

image.png

调整过期时间

1
2
3
4
5
6
7
# 调整单个topic的过期时间
kafka-configs.sh --bootstrap-server 1.1.1.1:9092 \
--entity-type topics --entity-name pod-logs \
--alter --add-config retention.ms=259200000

# 查看某个topic的过期时间
kafka-configs.sh --bootstrap-server 1.1.1.1:9092 --entity-type topics --entity-name pod-logs --describe