常用脚本
kafka执行脚本默认在安装的bin目录下，本文中示例均基于bin目录执行。
#查询topic状态，新建，删除，扩容
kafka-topics.sh
#查看，修改kafka配置
kafka-configs.sh
#配置，查看kafka集群鉴权信息
kafka-acls.sh
#生产命令
kafka-console-producer.sh
# 消费命令
kafka-console-consumer.sh
#查看消费者组，重置消费位点等
kafka-consumer-groups.sh
#kafka自带消费性能测试命令
kafka-consumer-perf-test.sh
# #kafka集群间同步命令
kafka-mirror-maker.sh
#重新选举topic分区leader
kafka-preferred-replica-election.sh
#kafka自带生产性能测试命令
kafka-producer-perf-test.sh
#kafka数据重平衡命令
kafka-reassign-partitions.sh
#kafka执行脚本
kafka-run-class.sh
#进程启动
kafka-server-start.sh
#进程停止
kafka-server-stop.sh
topic相关
使用kafka-topics.sh。
创建
./kafka-topics.sh --create --bootstrap-server 192.168.1.12:9092 --replication-factor 3 --partitions 2 --topic my-test-topic
参数说明：
–create：创建主题；
–bootstrap-server：指定kafka服务器地址；
–replication-factor：副本数量,注意不能大于broker数量;如果不提供,则会用集群中默认配置；
–partitions：分区数量,当创建或者修改topic的时候,用这个来指定分区数;如果创建的时候没有提供参数,则用集群中默认值; 注意如果是修改的时候,分区比之前小会有问题；
–topic <String: topic>：操作的topic名称；
–config <String: name=value>：操作配置项；
删除
./kafka-topics.sh --bootstrap-server 192.168.1.12:9092 --delete --topic my-test-topic
参数说明：
- –delete：删除主题；
修改
也可以理解为topic分区扩容。
# 修改单个topic
./kafka-topics.sh --bootstrap-server 192.168.1.12:9092 --alter --topic my-test-topic --partitions 6
# 批量修改topic
./kafka-topics.sh --topic ".*?" --bootstrap-server 192.168.1.12:9092 --alter --partitions 6
参数说明：
- –alter：修改主题；
查询描述
# 查询单个描述
./kafka-topics.sh --topic my-test-topic --bootstrap-server 192.168.1.12:9092 --describe --exclude-internal
# 查询所有描述
./kafka-topics.sh --topic ".*?" --bootstrap-server 192.168.1.12:9092 --describe --exclude-internal
参数说明：
- –describe：查看主题详细描述；
- –exclude-internal：排除kafka内部主题；
查询列表
# 查询所有主题
./kafka-topics.sh --bootstrap-server 192.168.1.12:9092 --list --exclude-internal
# 查询特定主题
./kafka-topics.sh --bootstrap-server 192.168.1.12:9092 --list --exclude-internal --topic "my-test-topic*"
参数说明：
- –list：查看所有主题；
config相关
使用kafka-configs.sh。
查询
# 查询单个topic配置
./kafka-configs.sh --describe --bootstrap-server 192.168.1.12:9092 --topic my-test-topic
# 查询所有topic动态配置
./kafka-configs.sh --describe --bootstrap-server 192.168.1.12:9092 --entity-type topics
# 查询所有topic动态+静态配置
./kafka-configs.sh --describe --bootstrap-server 192.168.1.12:9092 --entity-type topics --all
# 查询版本信息
./kafka-configs.sh --describe --bootstrap-server 192.168.1.12:9092 --version
修改
# 添加配置(--add-config)
./kafka-configs.sh --bootstrap-server 192.168.1.12:9092 --alter --entity-type topics --entity-name my-test-topic --add-config file.delete.delay.ms=222222,retention.ms=999999
# 删除配置(--delete-config)
./kafka-configs.sh --bootstrap-server 192.168.1.12:9092 --alter --entity-type topics --entity-name my-test-topic --delete-config file.delete.delay.ms,retention.ms
参数说明：
- –add-config：添加配置
- –delete-config：删除配置
- –entity-type：选择类型
- –entity-name：类型名称
生产者
./kafka-console-producer.sh --bootstrap-server 192.168.1.12:9092 --topic my-test-topic
>hello kafka
>
可选参数说明：
- –bootstrap-server：要连接的服务器；
- –topic：接收消息的主题名称；
- –batch-size：单个批处理中发送的消息数；
- –compression-codec：压缩编解码器；
- –max-block-ms：在发送请求期间，生产者将阻止的最长时间；
- –max-memory-bytes：生产者用来缓冲等待发送到服务器的总内存；
- –max-partition-memory-bytes：为分区分配的缓冲区大小；
- –message-send-max-retries：最大的重试发送次数；
- –metadata-expiry-ms：强制更新元数据的时间阈值(ms)；
- –producer-property：将自定义属性传递给生成器的机制；
- –producer.config：生产者配置属性文件；
- –property：自定义消息读取器；
- –request-required-acks：生产者请求的确认方式；
- –request-timeout-ms：生产者请求的确认超时时间；
- –retry-backoff-ms：生产者重试前，刷新元数据的等待时间阈值；
- –socket-buffer-size：TCP接收缓冲大小；
- –timeout：消息排队异步等待处理的时间阈值；
- –sync：同步发送消息；
- –version：显示 Kafka 版本；
- –help：打印帮助信息；
消费者
# 消费特定topic
./kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --topic my-test-topic
# 读取所有数据
./kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --from-beginning --topic my-test-topic
参数说明：
- –group：指定消费者所属组的ID；
- –topic：被消费的topic；
- –partition：指定分区 ；
- –offset：执行消费的起始offset位置 ;
- –whitelist：正则表达式匹配topic；
- –consumer-property：将用户定义的属性以key=value的形式传递给使用者；
- –consumer.config：消费者配置属性文件；
- –property：初始化消息格式化程序的属性；
- –from-beginning：从存在的最早消息开始，而不是从最新消息开始。注意如果配置了客户端名称并且之前消费过，那就不会从头消费了；**
- –max-messages：消费的最大数据量；
- –skip-message-on-error：如果处理消息时出错，请跳过它而不是暂停；
压力测试
生产者
# 发送 5000000 条大小为 1KB 的消息到地址192.168.1.12:9092
./kafka-producer-perf-test.sh --topic my-test-topic --num-records 5000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=192.168.1.12:9092
参数说明：
- –num-records：发送的消息数；
- –record-size：一条消息的大小byte；
- –throughput：每秒消息最大吞吐量，-1表示不限制吞吐量；
- –producer-props：生产者配置, 例如k1=v1,k2=v2；
消费者
./kafka-consumer-perf-test.sh --bootstrap-server 192.168.1.12:9092,192.168.1.13:9092,192.168.1.14:9092 --topic my-test-topic --messages 5000000 --threads 32 --reporting-interval 10000 --show-detailed-stats
参数说明：
- –messages：消费的数量；
- –threads：消费线程数；
- –reporting-interval：每隔10s打印一次消费消息信息；
- –show-detailed-stats：展示详细状态信息；
- –fetch-size：单次获取数据的大小；
