Kafka实操命令

kafka版本：scala2.11，kafka1.1.0

kafka_2.11-1.1.0.jar

Kafka命令行操作

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

kafka-topics.sh --zookeeper hadoop111:2181/kafka --list

2）创建topic

 kafka-topics.sh --zookeeper hadoop111:2181/kafka --create --replication-factor 3 --partitions 1 --topic first

选项说明：

--topic 定义topic名

--replication-factor 定义副本数

--partitions 定义分区数

3）删除topic

kafka-topics.sh --zookeeper hadoop111:2181/kafka --delete --topic first

需要server.properties中设置delete.topic.enable=true否则只是标记删除。

4）发送消息

kafka-console-producer.sh --broker-list hadoop111:9092 --topic first

>hello world

5）消费消息

kafka-console-consumer.sh --bootstrap-server hadoop111:9092 --from-beginning --topic first

--from-beginning：会把主题中以往所有的数据都读取出来。

6）查看某个Topic的详情

kafka-topics.sh --zookeeper hadoop112:2181/kafka --describe --topic first

7）修改分区数

kafka-topics.sh --zookeeper hadoop112:2181/kafka --alter --topic first --partitions 6

8）查看消费组的消费情况

kafka-consumer-groups.sh --bootstrap-server hadoop111:9092 --describe --group group_name

9）调整Kafka的偏移量 -- 调节偏移量必须让消费组处于inactive状态

1. 增加/减少kafka当前偏移量

kafka-consumer-groups.sh --bootstrap-server hadoop111:9092 --reset-offsets --group group_name --topic topic_name --shift-by +10000 --execute

将group_name组下的topic_name的偏移量加10000

kafka-consumer-groups.sh --bootstrap-server hadoop111:9092 --reset-offsets --group group_name --topic topic_name --shift-by -10000 --execute

将group_name组下的topic_name的偏移量减10000

2. 调节kafka偏移量至最新位置

kafka-consumer-groups.sh --bootstrap-server hadoop111:9092 --reset-offsets --group group_name --topic topic_name --to-latest --execute

调节group下的所有topic的偏移量至最新位置

kafka-consumer-groups.sh --bootstrap-server hadoop111:9092 --reset-offsets --group group_name --all-topics --to-latest --execute

--to-earliest：把位移调整到分区当前最小位移

--to-latest：把位移调整到分区当前最新位移