【Linux】：Kafka基础命令

目录

环境简介

一、操作主题命令

  1.1.创建主题

  1.2.指定主题的分区数、副本数创建主题

  1.3.查看kafka中的主题列表信息

  1.4.查看kafka找那个特定主题的详细信息

  1.5.修改主题的分区数(只能从小往大改)

  1.6.删除主题

二、操作消息命令

  2.1.生产者向指定主题发送消息

  2.2.消费者监听指定消息(消费者每次启动都从主题中最开始的消息开始监听)

  2.3.消费者监听指定主题的消息(消费者每次启动都从最新的消息开始监听)

三、消费者组操作命令

  3.1.创建一下消费者监听消息，并将该消费者放在名为testgroup消费者组下

  3.2.查看消费者组列表

  3.3.查看消费者组的详细信息

四、kafka集群中操作消息的命令

  4.1.向集群中发送消息

  4.2.消费者消费消息(未指定消费组)

  4.3.消费者消费消息(指定消费组)

环境简介

   Linux内核：Centos7

   Kafka版本：3.5.1

   执行命令的目录位置：Kafka安装目录的bin目录下：/usr/local/kafka/bin/

一、操作主题命令

  1.1.创建主题

./kafka-topics.sh --bootstrap-server localhost:9092  --create  --topic mytesttopic

  1.2.指定主题的分区数、副本数创建主题

./kafka-topics.sh --bootstrap-server localhost:9092  --create   --replication-factor 1 --partitions 1 --topic mytesttopic

  1.3.查看kafka中的主题列表信息

./kafka-topics.sh --bootstrap-server localhost:9092 --list

  1.4.查看kafka找那个特定主题的详细信息

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytesttopic 

  1.5.修改主题的分区数(只能从小往大改)

./kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic mytesttopic --partitions 2

  1.6.删除主题

./kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytesttopic

二、操作消息命令

  2.1.生产者向指定主题发送消息

./kafka-console-producer.sh broker-list --bootstrap-server localhost:9092  --topic mytesttopic 
## 下面为提示的消息输入框
>send message 1
>send message 2

  2.2.消费者监听指定消息(消费者每次启动都从主题中最开始的消息开始监听)

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

  2.3.消费者监听指定主题的消息(消费者每次启动都从最新的消息开始监听)

        注意：该监听方式，如果消费者想监听到消息，那么需要在消费者监听成功后，再使用命令让生产者发送消息

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

三、消费者组操作命令

  3.1.创建一下消费者监听消息，并将该消费者放在名为testgroup消费者组下

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testgroup --topic testtopic0914

  3.2.查看消费者组列表

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

  3.3.查看消费者组的详细信息

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

    消费者组中的各项列表含义如下：

      GROUP：消费者组名称
      TOPIC：消费者组监听的消息
      PARTITON：使用的是第几个分区
      CURRENT-OFFSET：当前消息的偏移量
      LOG-END-OFFSET：消费的偏移量
      LAG：还有多少消费没有被消费，即积压的消息数
 3.4.删除消费者组

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group {消费组}

四、kafka集群中操作消息的命令

  4.1.向集群中发送消息

[root@localhost kafka_2.12-3.5.1]# ./bin/kafka-console-producer.sh  --broker-list 192.168.154.128:9092,192.168.154.128:9093,192.168.154.128:9094 --topic topic0921

  4.2.消费者消费消息(未指定消费组)

[root@localhost kafka_2.12-3.5.1]# ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.154.128:9092,192.168.154.128:9093,192.168.154.128:9094 --from-beginning --topic topic0921
hello -- 消费到的消息

  4.3.消费者消费消息(指定消费组)

[root@localhost kafka_2.12-3.5.1]# ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.154.128:9092,192.168.154.128:9093,192.168.154.128:9094 --from-beginning --consumer-property group.id=testGroup1 --topic  topic0921
hello  -- 消费到的消息

思考：为什么有消费者，还有消费者组。

生产者发送一条消息，只能被同一个消费者组下的一个消费者消费，换句话说，同一个消费者组下，只能有一个消费者能监听到同一个topic里的消息。

     同一个消费者组下的消费者们，监听同一个topic时，topic中的消息只能被其中一个消费者监听到。
     专一性：
        如果消息一开始被消费者组中的某个消费者监听到，那么在该消费者的正常存活期间，消息将会一直被该消费者监听到(不是轮询问题)
     容错性：
        如果一直能监听到topic中消息的消费者挂掉了，那么当前消费者组中，其他监听同topic的消费者会继续监听到topic中的后续消息

生产者发送一条消息，可以被多个消费者组监听到，但是每个消费者组中，只能有一个消费者能监听到