命令行 · JAVA

[TOC] # 文档官方文档：http://kafka.apache.org/documentation.html 中文文档: http://orchome.com/kafka/index # topic * 查看当前服务器中的所有topic ~~~ kafka-topics.sh --list --zookeeper master:2181 ~~~ * 查看topic状态 ~~~ kafka-topics.sh --describe --zookeeper localhost:2181 --topic order ~~~ * 创建topic ~~~ kafka-topics.sh --create --zookeeper master:2181,slave1:2181,slave2:2181 --replication-factor 1 --partitions 1 --topic order ~~~ ~~~ --replication-factor 1 #复制1份,冗余数 --partitions 1 #创建1个分区 --topic #主题 ~~~ * 删除topic ~~~ kafka-topics.sh --delete --zookeeper master:2181 --topic order ~~~ 需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。 # 发送消费消息 * 通过shell命令发送消息 ~~~ kafka-console-producer.sh --broker-list master:9092 --topic order ~~~ 注意这里的master:9092和zk没有关系,这边是指生产的消息放到哪个kafka集群上运行 **zk中存的是偏移量和broker的状态信息** * 通过shell消费消息(--from-beginning从最开始的位置消费数据) ~~~ kafka-console-consumer.sh --zookeeper master:2181 --topic order --from-beginning ~~~ # 查看消息 * 查看消费位置 ~~~ kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper master:2181 --group testGroup ~~~ * 查看某个Topic的详情 ~~~ kafka-topics.sh --topic order --describe --zookeeper master:2181 ~~~ # 修改分区 * 对分区数进行修改 ~~~ kafka-topics.sh --zookeeper master --alter --partitions 15 --topic utopic ~~~ # maven的包 ~~~  <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.2</version> </dependency>  <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-streams</artifactId> <version>0.11.0.2</version> </dependency>  <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.12</artifactId> <version>0.11.0.0</version> </dependency> ~~~