基础命令

1.创建topic

kafka-topics.sh --bootstrap-server ${kafkaAddress} --create --topic ${topicName} --partitions ${partipartions} --replication-factor ${replication}

2.查看topic详情

kafka-topics.sh --bootstrap-server ${kafkaAddress} --topic ${topicName} --describe

3.删除topic

kafka-topics.sh --bootstrap-server ${kafkaAddress} --delete--topic ${topicName} --partitions ${partitions} --replication-factor ${replication}

4.查看topic list

kafka-topics.sh --bootstrap-server ${kafkaAddress} --list

5.消费topic

kafka-console-consumer.sh --bootstrap-server ${kafkaAddress} --topic ${topicName} --from-beginning

6.查看kafka consumer group 消费情况

kafka-consumer-groups.sh --describe --bootstrap-server ${kafkaAddress} --group ${groupName}

7.已经创建的topic修改partipartions 和 replication-factor 数量

step:

a.修改partitions数量

kafka-topics.sh --bootstrap-server ${kafkaAddress} --topic ${topicName} --alter --partitions 4

b.创建increase-replication-factor.json in config,配置各分区replication-factor位置

{

  "version": 1,

  "partitions": [

    {

      "topic": "${topicName}",

      "partition": 0,

      "replicas": [

        0,

        1

      ]

    },

    {

      "topic": "${topicName}",

      "partition": 1,

      "replicas": [

        1,

        2

      ]

    },

    {

      "topic": "${topicName}",

      "partition": 2,

      "replicas": [

        2,

        3

      ]

    },

    {

      "topic": "${topicName}",

      "partition": 3,

      "replicas": [

        3,

        0

      ]

    }

  ]

}

c.更新replication-factor

kafka-reassign-partitions.sh --bootstrap-server ${kafkaAddress} --reassignment-json-file config/increase-replication-factor.json --execute

3.创建topic话题

bin/kafka-topics.sh --create --topic test --bootstrap-server 192.168.218.128:9092 -partitions 3 -replication-factor 1

查看所有topic话题

bin/kafka-topics.sh --list --bootstrap-server 192.168.218.128:9092

查看指定话题的详情

bin/kafka-topics.sh --bootstrap-server 192.168.230.128:9092 --describe --topic test

3.创建生产者

bin/kafka-console-producer.sh --broker-list 192.168.218.128:9092 --topic test

4.创建消费者

bin/kafka-console-consumer.sh --bootstrap-server 192.168.218.128:9092 --topic test --from-beginning

## 创建topic（4个分区，2个副本）

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

### kafka版本 >= 2.2

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

## 分区扩容

### kafka版本 < 2.2

bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic1 --partitions 2

### kafka版本 >= 2.2

bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic topic1 --partitions 2

## 删除topic

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