草庐IT

kafka常用命令

gblfy 2023-04-18 原文

文章目录

一、KAFKA启停命令
1. 前台启动

kafka 前台启动命令：

bin/kafka-server-start.sh config/server.properties
2. 后台启动

kafka 后台启动命令：
后台常驻方式，带上参数 -daemon，如：

bin/kafka-server-start.sh -daemon config/server.properties

或者

nohup bin/kafka-server-start.sh config/server.properties &

指定 JMX port 端口启动，指定 jmx，可以方便监控 Kafka 集群

JMX_PORT=9991 /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
3. 停止命令

kafka 停止命令：

bin/kafka-server-stop.sh
二、Topic 相关命令
2.1. 创建 Topic

参数 --topic 指定 Topic 名，–partitions 指定分区数，–replication-factor 指定备份(副本)数

创建名为 test_kafka_topic 的 Topic

bin/kafka-topics.sh -zookeeper localhost:2181 --create --partitions 5 --replication-factor 1 --topic test_kafka_topic

注意，如果配置文件 server.properties 指定了 Kafka 在 zookeeper 上的目录，则参数也要指定，否则会报无可用的 brokers（下面部分命令也有同样的情况），如：

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic test
2.2. 查询 Topic 列表

列出所有 Topic

bin/kafka-topics.sh --list --zookeeper localhost:2181
2.3. 查询 Topic 详情

查询 Topic 的详细信息

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test_kafka_topic

说明：如果未指定 topic 则输出所有 topic 的信息

2.4. 增加 Topic 的 partition 数
bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic test_kafka_topic --partitions 5
2.5. 查看 topic 指定分区 offset 的最大值或最小值

time 为 -1 时表示最大值，为 -2 时表示最小值：

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic test_kafka_topic --time -1 --broker-list 127.0.0.1:9092 --partitions 0
2.6. 删除Topic

删除名为 test_kafka_topic 的 Topic

bin/kafka-topics.sh --delete --zookeeper localhost:2181  --topic test_kafka_topic

说明：在${KAFKA_HOME}/config/server.properties中配置 delete.topic.enable 为 true，这样才能生效，删除指定的 topic主题

三、消息 相关命令
3.1. 发送消息

生产者发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_kafka_topic
3.2. 消费消息（从头开始）

消费者查询消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092  --from-beginning --topic test_kafka_topic
3.3. 消费消息（从尾开始）

从尾部开始取数据

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_kafka_topic --offset latest
3.4. 消费消息（从尾开始指定分区）

从尾部开始取数据，指定分区消费：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_kafka_topic --offset latest --partition 0
3.5. 消费消息（指定分区指定偏移量）

–partition 指定起始偏移量消费–offset：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_kafka_topic  --partition 0 --offset 100
3.6. 指定分组->消费消息

消费者消费消息（指定分组）
注意给客户端命名之后,如果之前有过消费，那么–from-beginning就不会再从头消费了

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092  --from-beginning --topic test_kafka_topic --group t1

说明：
–from-beginning：表示从头开始接收数据
–group：指定消费者组

3.7. 取指定个数
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_kafka_topic --offset latest --partition 0 --max-messages 1
四、消费者 Group
4.1. 指定 Group

指定分组从头开始消费消息（应该会指定偏移量）

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -group test_group --from-beginning
4.2. 消费者 Group 列表
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
4.3. 查看 Group 详情
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --describe

输出日志：

Consumer group 'test_group' has no active members.

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
test            0          5               5               0               -               -               -

# CURRENT-OFFSET: 当前消费者群组最近提交的 offset，也就是消费者分区里读取的当前位置
# LOG-END-OFFSET: 当前最高水位偏移量，也就是最近一个读取消息的偏移量，同时也是最近一个提交到集群的偏移量
# LAG：消费者的 CURRENT-OFFSET 与 broker 的 LOG-END-OFFSET 之间的差距
4.4. 删除 Group 中 Topic
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --topic test --delete
4.5. 删除 Group
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --delete
五、补充命令
5.1.平衡 leader
bin/kafka-preferred-replica-election.sh --bootstrap-server localhost:9092
5.2. 自带压测工具
bin/kafka-producer-perf-test.sh --topic test --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers=localhost:9092
用命kafkaspanclasstoken

有关kafka常用命令的更多相关文章

  1. 华为常用命令 - 2

    system-view进入系统视图quit退到系统视图sysname交换机命名vlan20创建vlan(进入vlan20)displayvlan显示vlanundovlan20删除vlan20displayvlan20显示vlan里的端口20Interfacee1/0/24进入端口24portlink-typeaccessvlan20把当前端口放入vlan20undoporte1/0/10删除当前VLAN端口10displaycurrent-configuration显示当前配置02配置交换机支持TELNETinterfacevlan1进入VLAN1ipaddress192.168.3.100

  2. kafka如何动态消费新增topic主题 - 2

    一、解决痛点使用spring-kafka客户端，每次新增topic主题，都需要硬编码客户端并重新发布服务，操作麻烦耗时长。kafkaListener虽可以支持通配符消费topic，缺点是并发数需要手动改并且重启服务。对于业务逻辑相似场景，创建新主题动态监听可以用kafka-batch-starter组件二、组件能力1、新增topic名称为：auto.topic1（由于配置spring.kafka.consumer.prefix为auto，因此只有auto前缀的topic，才会被组件动态监听。）2、应用输出日志，监听到新增auto.topic1，并初始化客户端(主题刷新间隔为10s)3、发新的消

  3. Zookeeper、Nacos、Dubbo、Kafka之间的关系 - 2

    1.Zookeeper　　Zookeeper是 ApacheHadoop 的子项目，是一个树型的目录服务，支持变更推送，适合作为Dubbo服务的注册中心，工业强度较高。　　Zookeeper的功能主要是它的树形节点来实现的。当有数据变化的时候或者节点过期的时候，会通过事件触发通知对应的客户端数据变化了，然后客户端再请求zookeeper获取最新数据，采用push-pull来做数据更新。服务注册和消费信息直接存储在zk树形节点上，集群下采用过半机制保证服务节点间一致性。　2.Nacos　　Nacos是 Alibaba 公司推出的开源工具，用于实现分布式系统的服务发现与配置管理。Nacos是Dub

  4. Spark Kafka流媒体 - 如何确定批次的末端 - 2

    我使用Kafka流媒体从KAFKA主题中消费。（KafkaDirect流）此主题中的数据每5分钟从另一个来源到达。现在，我需要处理每5分钟后到达的数据，并将其转换为SparkDataFrame。现在，流是数据的连续流。我的问题是，如何确定我已经完成了在Kafka主题中加载的第一组数据的阅读？（以便我可以将其转换为数据框架并开始我的工作）我知道我可以提及某个数字的批处理间隔（在JavastreamingContext中），但是即使那样，我也永远无法确定源将数据将数据推到主题的时间。欢迎任何建议。看答案如果我正确理解您的问题，您希望不创建批处理，直到阅读5分钟的所有数据。开箱即用的Spark不会提

  5. git常用命令 - 2

    gitclonehttp:www.git.com.cn........ 克隆git项目gitbranch 查看分支gitbranch-r查看远程分支gitpushorigin--delete分支名 删除远程分支tmpgitcheckout切换分支gitcheckout-b切换并创建分支gitcheckout-b分支名origin/分支名(如果远程分支已存在最好用此命令，在创建分支时会把远程分支最新代码一并拉下来，不会把原分支代码带过来)gitbranch-D删除分支gitpushorigin--delete分支名gitpush--set-upstreamorigin分支名 推送本地分支到远端g

  6. 流批一体计算引擎-4-[Flink]消费kafka实时数据 - 2

    Python3.6.9Flink1.15.2消费KafakaTopicPyFlink基础应用之kafka通过PyFlink作业处理Kafka数据1环境准备1.1启动kafka(1)启动zookeeperzkServer.shstart(2)启动kafkacd/usr/local/kafka/nohup./bin/kafka-server-start.sh./config/server.properties>>/tmp/kafkaoutput.log2>&1&或者./bin/kafka-server-start.sh-daemon./config/server0.properties(3)查看进

  7. javascript - 在 Web 浏览器上实时接收 Kafka 事件 - 2

    我们使用Kafka集群实时发送/接收消息。我们能够成功地向Kafka主题发布消息。现在我们希望在使用JavaScript的Web浏览器上运行的单页应用程序(SPA)上实时接收这些消息。是否可以直接从Kafka将消息推送到在任何著名浏览器上运行的最新版本的SPA？我找到了使用NodeJS实时接收消息的示例，但没有找到在Web浏览器上运行JavaScript的示例。 最佳答案 Kafka有Javascript客户端，但鉴于您的用例描述，我建议您在浏览器端javascript和Kafka之间使用REST或WebSocket代理。这将确保消

  8. go - 转义 json 对象以创建 kafka 轴突有效载荷 - 2

    我一直在尝试创建需要转义格式的json数据的负载。我可以序列化该对象，但不确定如何对同一对象进行双重转义？我是否应该双重编码(marshal)我的对象以便它逃脱它？Input:{"new":{"Id":"1","Class":"23"}}Expected:{\"new\":{\"Id\":\"1\",\"Class\":\"23\"}} 最佳答案 将最后一行更改为fmt.Printf("%q",string(b))-这会导致格式为“转义字符串”。(或者如果你想存储转义字符串，```fmt.Sprintf``)https://play

  9. json - Kafka - Json(最佳实践) - 2

    我需要将RESTAPI调用的输出推送到KAFKA。Restapi返回json输出，其中包含支持信息以及数据输出到json.RawMessagetypeResponsestruct{RequestIDstring`json:"requestId"`Successbool`json:"success"`NextPageTokenstring`json:"nextPageToken,omitempty"`MoreResultbool`json:"moreResult,omitempty"`Errors[]struct{Codestring`json:"code"`Messagestring`

  10. asynchronous - 具有异步 goroutines 的 Kafka 消费者 - 2

    我正在为我的消费者使用sarama(https://github.com/Shopify/sarama/)和Kafka0.8.0。这是我的代码的样子:consumerLoop:for{select{caseevent:=我正在使用缓冲channel(c.sem)来控制一次可以运行多少个processJobgoroutine。这就是我控制消费者的并发/速度的方式。我在使用这种方法时遇到的问题是，如果我需要更改并发性，我必须关闭使用者并重新启动它(channel缓冲区大小是一个命令行标志)。我记录了已处理的偏移量，我必须查看我的日志以确定处理了哪些偏移量以及我希望消费者从哪里恢复。我想要一

随机推荐