贵阳网站开发公司,2022年最火的新闻摘抄,手机版景区网站,微信公众号登录不上常用脚本
kafka执行脚本默认在安装的bin目录下#xff0c;本文中示例均基于bin目录执行。
#查询topic状态#xff0c;新建#xff0c;删除#xff0c;扩容
kafka-topics.sh #查看#xff0c;修改kafka配置
kafka-configs.sh #配置#xff0c;查看kafka集群鉴权信息
kaf…常用脚本
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: namevalue操作配置项
删除
./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.ms222222,retention.ms999999# 删除配置(--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-sizeTCP接收缓冲大小–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将用户定义的属性以keyvalue的形式传递给使用者–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.servers192.168.1.12:9092参数说明
–num-records发送的消息数–record-size一条消息的大小byte–throughput每秒消息最大吞吐量-1表示不限制吞吐量–producer-props生产者配置, 例如k1v1,k2v2
消费者
./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单次获取数据的大小
