前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
首页
学习
活动
专区
工具
TVP最新优惠活动
返回腾讯云官网
社区首页 >专栏 >进击消息中间件系列（三）：Kafka 中 shell 命令使用

进击消息中间件系列（三）：Kafka 中 shell 命令使用

作者头像
民工哥
发布于 2023-08-22 14:12:20
4250
发布于 2023-08-22 14:12:20
举报
文章被收录于专栏：民工哥技术之路

创建 kafka topic

代码语言：javascript
复制
kafka-topics.sh --zookeeper node01:2181 --create --topic t_Name --partitions 30 --replication-factor 2

注：partitions指定topic分区数，replication-factor指定topic每个分区的副本数。

partitions分区数:

partitions ：分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量。

虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。

分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数。

分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1，producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。分区数只可以增加不可以减少。

replication-factor副本

replication factor 控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。

如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

查看所有topic列表

代码语言：javascript
复制
kafka-topics.sh --zookeeper node01:2181 --list

查看指定topic的详细信息

代码语言：javascript
复制
kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name

创建一个topic并生产消息

代码语言：javascript
复制
bootstrap_server=hadoop001:9092,hadoop002:9092,hadoop003:9092
zookeeper_connect=hadoop001:2181,hadoop002:2181,hadoop003:2181/kafka
kafka-topics.sh --create --zookeeper $zookeeper_connect --replication-factor 2 --partitions 3 --topic test001
kafka-console-producer.sh --broker-list $bootstrap_server --topic test001 --producer.config $KAFKA_HOME/config/producer.properties
代码语言：javascript
复制
bootstrap_server   #定义的kafka的broker集群节点，端口号在配置文件里可以查看。  
zookeeper_connect  #定义的zookeeper集群节点，kafka是配置文件中指定的对topic进行操作的虚拟用户。

对指定的topic消费

代码语言：javascript
复制
bootstrap_server=hadoop001:9092,hadoop002:9092,hadoop003:9092
zookeeper_connect=hadoop001:2181,hadoop002:2181,hadoop003:2181/kafka
kafka-console-consumer.sh --bootstrap-server $bootstrap_server --from-beginning --topic test001 --consumer.config $KAFKA_HOME/config/consumer.properties

查看topic某分区偏移量最大(小)值

代码语言：javascript
复制
kafka-run-class.sh kafka.tools.GetOffsetShell --topic t_Name --time -1 --broker-list node86:9092 --partitions 0

增加topic分区数

代码语言：javascript
复制
kafka-topics.sh --zookeeper node01:2181 --alter --topic t_cdr --partitions 10

查看topic消费进度

这个会显示出consumer group的offset情况， 必须参数为–group，不指定–topic，默认为所有topic。

代码语言：javascript
复制
Displays the: Consumer Group, Topic, Partitions, Offset, logSize, Lag, Owner for the specified set of Topics and Consumer Group
bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker
required argument: [group]
Option Description
------ -----------
--broker-info Print broker info
--group Consumer group.
--help Print this message.
--topic Comma-separated list of consumer
  topics (all topics if absent).
--zkconnect ZooKeeper connect string. (default: localhost:2181)
Example,
kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv
Group           Topic              Pid Offset   logSize    Lag    Owner
pv              page_visits        0   21       21         0      none
pv              page_visits        1   19       19         0      none
pv              page_visits        2   20       20         0      none

删除指定的topic

首先在kafka配置文件server.properties添加配置项 delete.topic.enable=true 之后执行命令：kafka-topics.sh --delete --zookeeper 【zookeeper server】 --topic 【topic name】

kafka数据迁移

方法一：通过增加partition数量
代码语言：javascript
复制
kafka-topics.sh --zookeeper localhost:2181 --alter --topic demo1 --partitions 3
方法二：通过重新分配partition

1.编写分配脚本并执行分配计划

代码语言：javascript
复制
vi ~/kafka_add_replicas.json

{"topics":
 [{"topic":"prod_log_simul"}],
 "version": 1
 }

之后运行分配计划

代码语言：javascript
复制
kafka-reassign-partitions.sh --zookeeper hdc-data1,hdc-data2,hdc-data3:2181 --topics-to-move-json-file ~/kafka_add_replicas.json --broker-list "0,1,2" --generate
代码语言：javascript
复制
–topics-to-move-json-file：#自己编写的分配脚本  
–broker-list：#中间写的是kafka broker集群的编号，例如：“125,127,10”

2.根据上一步生成的分配计划复制并配置json文件topic-reassignment.json，进行topic的重新分配

代码语言：javascript
复制
vi ~/topic-reassignment.json

之后把Proposed partition reassignment configuration #建议的分区配置 {“version”:1,“partitions”:[{“topic”:“event_request”,“partition”:0,“replicas”:[6,5]},{“topic”:“event_request”,“partition”:1,“replicas”:[7,6]}]} 的内容复制到文件里，然后根据自己的需要修改。

3.执行分配

代码语言：javascript
复制
kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --execute

执行前的分区分布

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name
Topic:event_request        PartitionCount:2   ReplicationFactor:2        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 3,4 Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 4,5 Isr: 4,5

执行后的分区分布

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name
Topic:event_request        PartitionCount:2   ReplicationFactor:4        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 6,5,3,4     Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 7,6,4,5     Isr: 4,5

4.检查分配的状态

查看分配的状态：正在进行

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [event_request,0] is still in progress
Reassignment of partition [event_request,1] is still in progress

查看“is still in progress” 状态时的分区，副本分布状态：发现Replicas有4个哦，说明在重新分配的过程中新旧的副本都在进行工作。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name       
        PartitionCount:2   ReplicationFactor:4        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 6,5,3,4     Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 7,6,4,5     Isr: 4,5

查看分配的状态：分配完成。

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [event_request,0] completed successfully
Reassignment of partition [event_request,1] completed successfully

查看“completed successfully”状态的分区，副本状态：已经按照生成的分配计划正确的完成了分区的重新分配。

代码语言：javascript
复制
[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name Topic:event_request        PartitionCount:2   ReplicationFactor:2        Configs:
        Topic: event_request      Partition: 0       Leader: 6  Replicas: 6,5 Isr: 6,5
        Topic: event_request      Partition: 1       Leader: 7  Replicas: 7,6 Isr: 6,7

kafka-configs.sh的使用–修改参数配置项

脚本语法解析

kafka-configs.sh参数解析

增加配置项

某个topic配置对象

代码语言：javascript
复制
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'k1=v1, k2=v2, k3=v3'

所有clientId的配置对象

代码语言：javascript
复制
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type clients --entity-default --add-config 'k1=v1, k2=v2, k3=v3'
例子
代码语言：javascript
复制
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'max.message.bytes=50000000, flush.messages=50000, flush.ms=5000'
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'max.message.bytes=50000000' --add-config 'flush.messages=50000'
删除配置项
代码语言：javascript
复制
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type clients --entity-name clientId --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-name $brokerId --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-default --delete-config ‘k1,k2,k3’
例子
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name test-cqy --delete-config 'segment.bytes'
修改配置项

修改配置项与增加语法格式相同，相同参数后端直接覆盖，列出entity配置描述。

代码语言：javascript
复制
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --entity-type topics --entity-name topicName --describe
bin/kafka-configs.sh--bootstrap-server localhost:9092 --entity-type brokers --entity-name $brokerId --describe
bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-default --describe
bin/kafka-configs.sh  --zookeeper localhost:2181/kafkacluster --entity-type users --entity-name user1 --entity-type clients --entity-name clientA --describe

kafka调整分区副本数

创建配置文件 increase-replication-factor.json， 这个配置文件说明，

代码语言：javascript
复制
{
    "partitions": [{
            "topic": "ba_spam_content",
            "partition": 0,
            "replicas": [1, 2]
        },
        {
            "topic": "ba_spam_content",
            "partition": 1,
            "replicas": [0, 2]
        },
        {
            "topic": "ba_spam_content",
            "partition": 2,
            "replicas": [1, 0]
        }
    ],
    "version": 1
}

可以看到，单机的情况，replicas 只在0这个broke上。

开始设置副本数

代码语言：javascript
复制
./bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file  increase-replication-factor.json --execute

查看副本情况

代码语言：javascript
复制
./bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file  increase-replication-factor.json --verify

查看分区信息

代码语言：javascript
复制
kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic ba_spam_content

验证设置的结果情况，可以看到，我们的结果非常好，都成功了

后面可以看到配置很OK，看到有备份信息了。

bin/kafka自带脚本解释

kafka消费者组查看

不同版本可能命令稍微有不同，根据提示配置相关参数。使用脚本：./kafka-consumer-groups.sh 用于查询偏移量信息，数据积压等信息。

代码语言：javascript
复制
kafka消费者组分两种情况，分别查看，根据版本由0.9为版本划分
0.9之前偏移量由zookeepr管理，由于zk不适合大批量写操作，
0.9之后由kafka自己管理，偏移量信息都保存在__consumer_offsets的topic中
查看zk管理的消费者组

This will only show information about consumers that use ZooKeeper (not those using the Java consumer API) 这将只显示使用ZooKeeper的消费者的信息(而不是使用Java消费者API的消费者)。

代码语言：javascript
复制
./kafka-consumer-groups.sh --new-consumer --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --list
./kafka-consumer-groups.sh --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --list
查看kafka管理的消费者组

仅仅查看由java 的API开发的consumer

代码语言：javascript
复制
./kafka-consumer-groups.sh --new-consumer --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --list
查看特定consumer group 消费情况

同样根据新/旧版本的consumer，分别指定bootstrap-server与zookeeper参数:

代码语言：javascript
复制
#zk管理的消费者组查看
./kafka-consumer-groups.sh --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --group t-1810-1 --describe
#kafka管理的消费者组查看
./kafka-consumer-groups.sh --new-consumer --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --group test --describe

查询结果：
LAG:积压的偏移量
LOG-END-OFFSET:接受到的最大偏移
CURRENT:已消费的最大偏移

使用 kafka-run-class.sh查看topic

查看消费进度

和上面查看结果一样。只是方式稍微不同：通过指定类去查看消费者组的详细消费kafka.admin.ConsumerGroupCommand类。

代码语言：javascript
复制
#使用规则：由于新旧版本不同，指定不同的zk,boot去查看消费者组
 ./kafka-run-class.sh
 0.9版本以后类 ：kafka.admin.ConsumerGroupCommand
 0.9版本以前类：kafka.tools.ConsumerOffsetChecker 0.9之后该类貌似已经不存在
  --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 ：#查看zk管理的消费者组
  --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 ：#查看kafka自己管理的消费者组
   --group  t-1810-1 
   --describe

#用例：
 ./kafka-run-class.sh kafka.admin.ConsumerGroupCommand --zookeeper hdp01:2181,hdp02:2181,hdp03:2181
  --group  t-1810-1 --describe
./kafka-run-class.sh kafka.admin.ConsumerGroupCommand --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --group  t-1810-1 --describe
查看topic最大消息数
代码语言：javascript
复制
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 194.168.221.17:21005,194.168.221.18:21005,194.168.221.19:21005,194.168.221.20:21005 --topic GQ_RESOURCE --time -1 

-1:#表示分区当前生产的最大消息条。总消息数=各分区和
3:#解码存储文件  
   可以用来解码存储的.log和.index文件  
.sh kafka.tools.DumpLogSegments --files name.log – print-data-log

kafka一键启动/关闭脚本

启动脚本
代码语言：javascript
复制
#!/bin/bash
brokers="hdp01 hdp02 hdp03"
KAFKA_HOME="/home/hadoop/apps/kafka_2.11-1.1.0"
KAFKA_NAME="kafka_2.11-1.1.0"

echo "INFO : Begin to start kafka cluster ..."

for broker in $brokers
do
  echo "INFO : Starting ${KAFKA_NAME} on ${broker} ..."
  ssh ${broker} -C "source /etc/profile; sh ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties 1>~/logs/kafka_std.log 2>~/logs/kafka_err.log &"
  if [[ $? -eq 0 ]]; then
      echo "INFO:[${broker}] Start successfully"
  fi
done
echo "INFO:Kafka cluster starts successfully !"
停止脚本
代码语言：javascript
复制
#!/bin/bash
brokers="hdp01 hdp02 hdp03"
KAFKA_HOME="/home/hadoop/apps/kafka_2.11-1.1.0"
KAFKA_NAME="kafka_2.11-1.1.0"

echo "INFO : Shut down kafka cluster ..."

for broker in $brokers
do
  echo "INFO : Starting ${KAFKA_NAME} on ${broker} ..."
  ssh ${broker} -C "source /etc/profile; sh ${KAFKA_HOME}/bin/kafka-server-stop.sh"
  if [[ $? -eq 0 ]]; then
      echo "INFO:[${broker}] stop successfully"
  fi
done
echo "INFO:Kafka cluster stop successfully !"

参考文章：https://blog.csdn.net/weixin_43930865/article /details/102762677 https://blog.csdn.net/qq_43536701/ article/details/111150285

本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2023-07-28，如有侵权请联系 cloudcommunity@tencent.com 删除
shell
kafka
脚本
配置
消息中间件

本文分享自 民工哥技术之路 微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ，欢迎热爱写作的你一起参与！

shell
kafka
脚本
配置
消息中间件
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
LV.
文章
0
获赞
0
目录
  • partitions分区数:
  • 查看所有topic列表
  • 查看指定topic的详细信息
  • 创建一个topic并生产消息
  • 对指定的topic消费
  • 查看topic某分区偏移量最大(小)值
  • 查看topic消费进度
  • 删除指定的topic
  • kafka数据迁移
    • 方法一：通过增加partition数量
      • 方法二：通过重新分配partition
      • kafka-configs.sh的使用–修改参数配置项
        • 脚本语法解析
          • 例子
          • bin/kafka自带脚本解释
            • kafka消费者组查看
              • 查看特定consumer group 消费情况
              • kafka一键启动/关闭脚本
                • 启动脚本
                  • 停止脚本
                  相关产品与服务
                  消息队列 TDMQ
                  消息队列 TDMQ （Tencent Distributed Message Queue）是腾讯基于 Apache Pulsar 自研的一个云原生消息中间件系列，其中包含兼容Pulsar、RabbitMQ、RocketMQ 等协议的消息队列子产品，得益于其底层计算与存储分离的架构，TDMQ 具备良好的弹性伸缩以及故障恢复能力。
                  产品介绍产品文档
                  精选特惠 用云无忧
                  领券

                  腾讯云开发者

                  扫码关注腾讯云开发者

                  扫码关注腾讯云开发者

                  领取腾讯云代金券

                  热门产品

                  热门推荐

                  更多推荐

                  Copyright © 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

                  深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

                  腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

                  问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

                  Copyright © 2013 - 2024 Tencent Cloud.

                  All Rights Reserved. 腾讯云 版权所有

                  登录 后参与评论