Topic相关操作命令

这个是从我本地word文档直接粘贴过来的, 以前是给自己看的,可能排版有点乱,等有时间我优化优化.

（一）主题操作

kafka-topics.sh 主题相关的操作

1．创建主题

主题的信息属于Kafka的元数据,是存在Zookeeper里面的brokers/topics里面,元数据是以永久节点的方式来保存的.
注意：①.partition可以有多个 ②副本的数量不能大于broker的数量

方式1

kafka-topics.sh —zookeeper zjj101:2181 —create —topic hello —partitions 2 —replication-factor 2

说明
-create —topic 固定的,
—partitions 指定分区有几个,必填
—replication-factor 指定副本数量,必填, 每个broker最多存储一份副本，所以在创建主题时，副本数不能超过当前可用的broker的数量！

这种方式 kafka集群根据负载均衡的策略，自动将分区分配到对应的broker实例中.

方式2

明确告诉kafka，创建多少个分区，以及每个分区的副本到底选择哪个broker

kafka-topics.sh —zookeeper zjj101:2181 —create —topic hello2 —replica-assignment 101:102,102:103

多个分区逗号间隔.

在Zookeeper查看创建的主题

zookeeper的hello这个永久节点显示:

这个意思是 一号区有两个副本,分别在102和103副本. 零号区也有两个副本,分别在101和102上面

2．查询所有的主题

kafka-topics.sh —zookeeper zjj101:2181 —list

3．查看某个主题详细信息

kafka-topics.sh —zookeeper zjj101:2181 —describe —topic hello

Topic: hello Partition: 0 Leader: 101 Replicas: 101,102 Isr: 101,102

意思是0号副本的 Leader是101分区 副本是101和102

4．修改主题

只能修改分区数(只允许增加)和副本的放置策略

bin/kafka-topics.sh —zookeeper zjj101:2181 —alter —topic hello2 —partitions 3

5．删除主题

这个删除只是给Zookeeper里面的元数据删除,但是实际数据并没有删除,它是标记删除,会给实际数据改个名字,然后会有个删除线程定时扫描删除,默认是一天删除.

bin/kafka-topics.sh —zookeeper hadoop102:2181 —delete —topic hello1

6. 修改分区数

kafka-topics.sh —zookeeper hadoop102:2181 —alter —topic first —partitions 6

（二）生产者消费者

1．创建启动生产者

kafka提供了用于测试的producer,位置在kafka_2.11-0.11.0.2/bin目录下面,有个kafka-console-producer.sh

命令:

kafka-console-producer.sh —broker-list zjj101:9092 —topic ods_order_info

2．创建启动消费者

kafka_2.11-0.11.0.2/bin目录下面,有个kafka-console-consumer.sh 将生产者的数据打印在控制台上面

命令 : kafka-console-consumer.sh —bootstrap-server zjj101:9092 —topic hello

输入完了之后就卡着了.. 等待生产者生产数据.

3. 指定消费的partition

kafka-console-consumer.sh —bootstrap-server hadoop102:9092 —topic aa2 —from-beginning —partition 1
注意 ：
1.如果不指明partition那么所有的partition都会消费
2.—from-beginning ： 从头开始消费
（如果指明了partition那么就是从这个partition的头消费这一个partition）


kafka-console-consumer.sh —bootstrap-server hadoop102:9092
—topic aa2 —partition 0 —offset earliest

4. 创建消费者组

kafka-console-consumer.sh —bootstrap-server hadoop102:2181 —topic aa2 —consumer-property group.id=组名

注意：只要组名一样就是同一个消费者组