1、Topic相关:创建Topic、删除Topic、查看Topic详细信息、查看Topic列表、修改topic分区数
创建Topic

 
  1. 创建一个3分区1副本名为test的topic,必须指定分区数 --partitions 和副本数--replication-factor,其中副本数量不能超过kafka节点(broker)数量

  2. ./kafka-topics.sh --zookeeper localhost:2181 --topic test --partitions 3 --replication-factor 1 --create

  • 删除Topic
 
  1. # 删除名为test的topic

  2. # 删除topic时只有在kafka安装目录config目录下的server.properties中将delete.topic.enable 设置为true topic才会真实删除,否则只是标记为删除,实则不会删除

  3. ./kafka-topics.sh --zookeeper localhost:2181 --topic test --delete

查看某个Topic 分区 副本信息

 
  1. # 查看名为test的topic的详细信息,分区 副本的数量

  2. ./kafka-topics.sh --zookeeper localhost:2181 --topic test --describe

  •  查看有那些Topic
 
  1. # 查看kafka中创建了那些topic

  2. ./kafka-topics.sh --zookeeper localhost:2181 --list

修改topic的分区数

 
  1. # 将名为test的topic 修改为4个分区

  2. # 注意 分区数只能增加不能减少

  3. ./kafka-topics.sh --zookeeper localhost:2181 -alter --partitions 4 --topic test

2、生产者相关:往某个topic中生产数据
使用命令行往某个topic中写入数据

 
  1. # 使用命令行 给名为 test 的topic 中生产数据

  2. # 执行以下命令,然后在命令行中写入要发送kafka的数据回车即可发送数据到kafka

  3. ./kafka-console-producer.sh --broker-list localhost:9092 --topic test

3、消费者相关:从某个topic中消费数据
消费某个topic中的最新数据

 
  1. # 0.8版本及以下的的kafka 使用如下命令test topic中的数据

  2. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

  3. # 指定消费10条数据

  4. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --max-messages 10

  5. # 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费

  6. ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

消费某个topic中最老的数据

 
  1. # 0.8版本及以下的的kafka 使用如下命令test topic中的数据

  2. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

  3. # 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费

  4. ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

消费某个topic中的数据并指定groupid

 
  1. # 在命令行消费某个topic中的数据通过/config/consumer.properties 配置文件指定groupid

  2. # 0.8版本及以下的的kafka 使用如下命令test topic中的数据

  3. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties

  4. # 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费

  5. ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties

将/config/consumer.properties配置文件中groupid对应的offset删除,该groupid重置为未使用状态

 
  1. # 使用这条命令会从最新消息开始消费,会将之前groupid记录的offset重置,并重新开始记录

  2. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties --delete-consumer-offsets

  3. # 使用consumer.properties 不可以和--from-beginning一同使用 除非与--delete-consumer-offsets一同使用

  4. # 使用这条命令会从头开始消费数据,会将之前groupid记录的offset重置,并重新开始记录

  5. ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties --delete-consumer-offsets --from beginning

4、消费组(group)相关:查看消费者group、查看消费者消费情况(消费至那个offset/积压数据量多少)
查看有那些消费者group

 
  1. # 0.8版本及以下的的kafka 使用如下命令查看有那些消费者group

  2. ./kafka-consumer-groups.sh --zookeeper localhost:2181 --list

  3. # 0.9版本及以上的kafka建议使用如下命令查看有那些消费者group,当然也可使用上一条命令消费

  4. ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看某个消费者消费情况(消息队列堆积情况)

 
  1. # 0.8版本及以下的的kafka 使用如下命令查看名为testgroup 的消费组的消费情况

  2. ./kafka-consumer-groups.sh --zookeeper localhost:2181 --group testgroup --describe

  3. # 0.9版本及以上的的kafka 使用如下命令查看名为testgroup 的消费组的消费情况

  4. ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group testgroup --describe

5、修改某个消费组的偏移量(offset)
通过zk客户端对topic的分区修改offset 为任意偏移量

 
  1. # 独立安装的zk,进入zookeeper安装目录的bin目录下,使用如下命令进入zk客户端

  2. ./zkCli.sh -server localhost:2181

  3. # 非独立安装的的zk, 直接在kafka安装目录bin目录下,使用如下命令进入zk客户端

  4. ./zookeeper-shell.sh localhost:2181

  5. # 进入zk客户端后可查看某个分区的偏移量 例如名为test的topic的消费者组 test-consumer-group 0分区的offset的消费情况

  6. get /consumers/test-consumer-group/offsets/test/0

  7. # 设置名为test的topic的消费者组 test-consumer-group 0分区的offset 为1000

  8. set /consumers/test-consumer-group/offsets/test/0 1000

通过kafka内置的kafka.tools.UpdateOffsetsInZK类实现修改某个topic 的消费组(config/consumer.properties中配置的groupid)的所有分区的偏移量为最新(latest)或者最旧(earliest)

 
  1. # 将名为test的topic的消费组(groupid必须从consumer.properties获取,即需要将需要修改的groupid写入consumer.properties配置文件)所有分区的offset设置为最早earliest

  2. ./kafka-run-class.sh kafka.tools.UpdateOffsetsInZK earliest ../config/consumer.properties test

  3. # 将名为test的topic的消费组(groupid必须从consumer.properties获取,即需要将需要修改的groupid写入consumer.properties配置文件)所有分区的offset设置为最新latest

  4. ./kafka-run-class.sh kafka.tools.UpdateOffsetsInZK latest ../config/consumer.properties test

0.11.0.0及以上版本修改偏移量可使用Kafka自带的kafka-consumer-groups.sh脚本

 
  1. # 以下可将--zookeeper localhost:2181 更换为--bootstrap-server localhost:9092 高版本的消费者建议连接bootstrap

  2. # 将test topic的消费组test-consumer-group的0分区的偏移量设置为最新

  3. ./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test:0 --reset-offsets --to-earliest –execute

  4. # 将test topic的消费组test-consumer-group的0和1分区的偏移量设置为最旧

  5. ./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test:0,1 --reset-offsets --to-latest –execute

 
  1. # 将test topic的消费组test-consumer-group的所有分区的偏移量设置为1000

  2. ./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test --reset-offsets --to-offset 1000 –execute

  3. # --reset-offsets后可以跟的其他用法:--to-current:把位移调整到分区当前位移

  4. # --reset-offsets后可以跟的其他用法:--shift-by N: 把位移调整到当前位移 + N处,注意N可以是负数,表示向前移动

  5. # --reset-offsets后可以跟的其他用法:--to-datetime <datetime>:把位移调整到大于给定时间的最早位移处,datetime格式是yyyy-MM-ddTHH:mm:ss.xxx,比如2017-08-04T00:00:00.000

Logo

华为云1024程序员节送福利,参与活动赢单人4000元礼包,更有热门技术干货免费学习

更多推荐