kafka相关操作

记录下kafka相关的命令。

查看topic

1	/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server kafka_ip:kafka_port

创建topic

1	/opt/kafka/bin/kafka-topics.sh --create --bootstrap-server kafka_ip:kafka_port --replication-factor 2 --partitions 1 --topic topic_name

删除topic

1	/opt/kafka/bin/kafka-topics.sh --delete --bootstrap-server kafka_ip:kafka_port --topic topic_name

查看topic 详细信息

1	/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka_ip:kafka_port --describe --topic topic_name

修改topic副本数

修改前一个副本：

编写修改副本计划

{"version":1,
"partitions":[
{"topic":"uat.qifu_saas_owms.tables","partition":0,"replicas":[0,1]}
]
}

注：如果topic有多个分区，每个分区都要写上，replicas后面接的是broker_id，想加几个副本就在后面写上对应的id，删除副本就把对应的id去掉，如：

{"version":1,
"partitions":[
{"topic":"test","partition":0,"replicas":[0,1]},
{"topic":"test","partition":1,"replicas":[0,1]},
{"topic":"test","partition":2,"replicas":[0,1]},
{"topic":"test","partition":3,"replicas":[0,1]},
{"topic":"test","partition":4,"replicas":[0,1]},
{"topic":"test","partition":5,"replicas":[0,1]},
{"topic":"test","partition":6,"replicas":[0,1]}
]
}

执行修改：

1	/opt/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server kafka_ip:kafka_port --reassignment-json-file replication.json --execute

修改后副本数：

修改分区数据（只能增加）

1	/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka_ip:kafka_port --alter --topic test --partitions 4

创建生产者

1	/opt/kafka/bin/kafka-console-producer.sh --broker-list kafka_ip:kafka_port --topic test

创建消费者

1	/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka_ip:kafka_port --topic test --from-beginning

创建消费者组

1	/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka_ip:kafka_port --topic test --consumer.config /opt/kafka/config/consumer.properties

列出消费者群组

1	/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka_ip:kafka_port --list

查看消息积压

1	/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka_ip:kafka_port --group owms-consumer3 --describe