kafka 常用命令

查看当前topic列表

bin/kafka-topics.sh --list --zookeeper ip:port,ip:port,ip:port

 

查看某topic

bin/kafka-topics.sh --describe --topic topic_test --zookeeper ip:port,ip:port,ip:port

 

创建topic

bin/kafka-topics.sh --create --zookeeper ip:port,ip:port,ip:port --replication-factor 3 --partitions 3 --topic topic_test

 

赋予producer和consumer权限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=ip:port,ip:port,ip:port --add --allow-principal User:* --producer --topic topic_test

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=ip:port,ip:port,ip:port --add --allow-principal User:* --consumer --topic topic_test --group '*'

 

赋予cluster权限

 

显示某个topic的acl权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ip:port,ip:port,ip:port --list --topic topic_test

 

单个pub/sub

bin/kafka-console-producer.sh --broker-list ip:port,ip:port,ip:port --topic topic_test --producer.config config/producer.properties

bin/kafka-console-consumer.sh --bootstrap-server ip:port,ip:port,ip:port --topic topic_test --consumer.config config/consumer.properties --from-beginning

 

查看是否有under-replicated-partitions

bin/kafka-topics.sh --zookeeper ip:port,ip:port,ip:port --describe --under-replicated-partitions

 

批量发送消息

bin/kafka-producer-perf-test.sh --num-records 10000 --record-size 256 --throughput 10 --topic topic_test --producer-props acks=-1 bootstrap.servers=ip:port,ip:port,ip:port --producer.config config/producer.properties --print-metrics

bin/kafka-consumer-perf-test.sh --broker-list ip:port,ip:port,ip:port --messages 10000 --threads 1 --topic topic_test --consumer.config config/consumer.properties --print-metrics

 

启动kafka

bin/kafka-server-start.sh -daemon /config/server.properties

 

停止kafka

bin/kafka-server-stop.sh

 

leader rebalance

bin/kafka-preferred-replica-election.sh --zookeeper ip:port,ip:port,ip:port