kafka（二）——常用命令_网站优化分享

常用脚本

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 ：操作的topic名称；
–config ：操作配置项；

删除
```
./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：单次获取数据的大小；
        
        广州网页设计培训班可以制作网页的软件房地产网站制作网页改版要怎么改南京网站制作专业公司网站建设北京

上一篇：13个常见的 WordPress 块编辑器问题以及如何修复它们

下一篇：Kafka为什么在消息积压时不能直接通过消费者水平扩容来提升消费速度?

常用脚本

topic相关

创建

删除

修改

查询描述

查询列表

config相关

查询

修改

生产者

消费者

压力测试

生产者

消费者