二、*Kafka命令行操作*
Kafka中提供了许多命令行工具(位于$KAFKA HOME/bin 目录下)用于管理集群的变更。
| kafka-configs.sh | 用于配置管理 |
|---|---|
| kafka-console-consumer.sh | 用于消费消息 |
| kafka-console-producer.sh | 用于生产消息 |
| kafka-console-perf-test.sh | 用于测试消费性能 |
| kafka-topics.sh | 用于管理主题 |
| kafka-dump-log.sh | 用于查看日志内容 |
| kafka-server-stop.sh | 用于关闭Kafka服务 |
| kafka-preferred-replica-election.sh | 用于优先副本的选举 |
| kafka-server-start.sh | 用于启动Kafka服务 |
| kafka-producer-perf-test.sh | 用于测试生产性能 |
| kafka-reassign-partitions.sh | 用于分区重分配 |
\1. 创建topic
kafka-topics.sh –create –topic test1 –partitions 1 –replication-factor 2 –zookeeper node1:2181
\2. 删除topic
kafka-topics.sh –delete –topic tpc1 –zookeeper node1:2181
\3. 查看topic
kafka-topics.sh –list –zookeeper node1:2181,node2:2181,node3:2181 _consumer_offsets
\4. 增加分区数
bin/kafka-topics.sh –alter –topic tpc_1 –partitions 3 –zookeeper node1:2181
Kafka 只支持增加分区,不支持减少分区
*原因是:减少分区,代价太大(数据的转移,日志段拼接合并)*
\5. 动态配置topic参数
–添加、修改配置参数（开启压缩发送传输种提高kafka消息吞吐量的有效办法(‘gzip’, ‘snappy’, ‘lz4’, ‘zstd’)）
bin/kafka-configs.sh –zookeeper node1:2181 –entity-type topics –entity-name tpc_1 –alter –add-config compression.type=gzip
–删除配置参数
bin/kafka-configs.sh –zookeeper node1:2181 –entity-type topics –entity-name tpc_1 –alter –delete-config compression.type
\6. Kafka命令行生产者与消费者操作
–生产者:kafka-console-producer
bin/kafka-console-producer.sh –broker-list node1:9092, node2:9092, node3:9092 –topic tpc_1
–消费者:kafka-console-consumer
bin/kafka-console-consumer.sh –bootstrap-server node1:9092, node2:9092, node1:9092 –topic tpc_1 –from-beginning
