1、集群管理 前台启动broker bin/kafka-server-start.sh <path>/server.propertiesCtrl + C 关闭 后台启动broker bin/kafka-server-start.sh -daemon <path>/server.properties关闭broker bin/kafka-server-stop.sh2、Topic管理创建topic bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 3 --replication-factor 3 --topic topicname 删除topic bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicname 查询topic列表 bin/kafka-topics.sh --zookeeper localhost:2181 --list查询topic详情 bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic topicname 修改topic bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 6 --topic topicname 3、Consumer-Groups管理查询消费者组 bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 查询消费者组详情 bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group groupname 重设消费者组位移 最早处bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-earliest --execute最新处bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-latest --execute某个位置bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-offset 2000 --execute调整到某个时间之后得最早位移bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-datetime 2019-09-15T00:00:00.000删除消费者组 bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --delete --group groupname4、脚本工具producer脚本 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topicname 参数含义:--compression-codec lz4 压缩类型--request-required-acks all acks的值--timeout 3000 linger.ms的值--message-send-max-retries 10 retries的值--max-partition-memory-bytes batch.size值consumer脚本 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning指定groupidbin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning--consumer-property group.id=old-consumer-group指定分区bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning--partition 0kafka-run-class脚本 kafka-run-class.sh kafka.tools.ConsoleConsumer 就是 kafka-console-consumer.shkafka-run-class.sh kafka.tools.ConsoleProducer 就是 kafka-console-producer.sh获取topic当前消息数 kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topicname --time -1--time -1表示最大位移 --time -2表示最早位移 查询_consumer_offsets bin/kafka-simple-consumer-shell.sh --topic _consumer_offsets --partition 12 --broker-list localhost:9092 --formatter "kafka.coorfinator.GroupMetadataManager\$OffsetsMessageFormatter"5、MirrorMaker跨机房灾备工具 bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB
|