Kafka命令介绍和使用以及案例_kafka应用案例

资料

所有脚本 

[root@localhost kafka]# ls bin/ -al
总用量 136
drwxr-xr-x. 3 root root 4096 3月  10 2019 .
drwxr-xr-x. 9 root root  134 8月  16 11:14 ..
-rwxr-xr-x. 1 root root 1421 3月  10 2019 connect-distributed.sh
-rwxr-xr-x. 1 root root 1418 3月  10 2019 connect-standalone.sh
-rwxr-xr-x. 1 root root  861 3月  10 2019 kafka-acls.sh
-rwxr-xr-x. 1 root root  873 3月  10 2019 kafka-broker-api-versions.sh
-rwxr-xr-x. 1 root root  864 3月  10 2019 kafka-configs.sh
-rwxr-xr-x. 1 root root  945 3月  10 2019 kafka-console-consumer.sh
-rwxr-xr-x. 1 root root  944 3月  10 2019 kafka-console-producer.sh
-rwxr-xr-x. 1 root root  871 3月  10 2019 kafka-consumer-groups.sh
-rwxr-xr-x. 1 root root  948 3月  10 2019 kafka-consumer-perf-test.sh
-rwxr-xr-x. 1 root root  871 3月  10 2019 kafka-delegation-tokens.sh
-rwxr-xr-x. 1 root root  869 3月  10 2019 kafka-delete-records.sh
-rwxr-xr-x. 1 root root  866 3月  10 2019 kafka-dump-log.sh
-rwxr-xr-x. 1 root root  863 3月  10 2019 kafka-log-dirs.sh
-rwxr-xr-x. 1 root root  862 3月  10 2019 kafka-mirror-maker.sh
-rwxr-xr-x. 1 root root  886 3月  10 2019 kafka-preferred-replica-election.sh
-rwxr-xr-x. 1 root root  959 3月  10 2019 kafka-producer-perf-test.sh
-rwxr-xr-x. 1 root root  874 3月  10 2019 kafka-reassign-partitions.sh
-rwxr-xr-x. 1 root root  874 3月  10 2019 kafka-replica-verification.sh
-rwxr-xr-x. 1 root root 9289 3月  10 2019 kafka-run-class.sh
-rwxr-xr-x. 1 root root 1376 3月  10 2019 kafka-server-start.sh
-rwxr-xr-x. 1 root root  997 3月  10 2019 kafka-server-stop.sh
-rwxr-xr-x. 1 root root  945 3月  10 2019 kafka-streams-application-reset.sh
-rwxr-xr-x. 1 root root  863 3月  10 2019 kafka-topics.sh
-rwxr-xr-x. 1 root root  958 3月  10 2019 kafka-verifiable-consumer.sh
-rwxr-xr-x. 1 root root  958 3月  10 2019 kafka-verifiable-producer.sh
-rwxr-xr-x. 1 root root 1722 3月  10 2019 trogdor.sh
drwxr-xr-x. 2 root root 4096 3月  10 2019 windows
-rwxr-xr-x. 1 root root  867 3月  10 2019 zookeeper-security-migration.sh
-rwxr-xr-x. 1 root root 1393 3月  10 2019 zookeeper-server-start.sh
-rwxr-xr-x. 1 root root 1001 3月  10 2019 zookeeper-server-stop.sh
-rwxr-xr-x. 1 root root  968 3月  10 2019 zookeeper-shell.sh

1. kafka版本

kafka没有提供命令或api直接差kafka版本。可以用如下方法：

[root@localhost kafka]# find ./libs/ -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'
kafka_2.11-2.2.0.jar

结果:
就可以看到kafka的具体版本了。
其中，2.11为scala版本，2.2.0为kafka版本。

export PATH=$ZOOKEEPER_HOME/bin:$PATH:/opt/kafa/kafka/bin

集群管理

# 启动zookeeper
zookeeper-server-start.sh config/zookeeper.properties &
 
 
# 停止zookeeper
zookeeper-server-stop.sh
 
 
# 前台启动broker Ctrl + C 关闭
kafka-server-start.sh <path>/server.properties
 
 
# 后台启动broker
kafka-server-start.sh -daemon <path>/server.properties
 
 
# 关闭broker
kafka-server-stop.sh

topic相关

kafka-topics.sh 脚本主要负责 topic 相关的操作。它的具体实现是通过kafka-run-class 来调用 TopicCommand 类，并根据参数执行指定的功能。

TopicCommand.createTopic() 方法负责创建 Topic，其核心逻辑是确定新建 Topic 中有多少个分区及每个分区中的副本如何分配，既支持使用 replica-assignment 参数手动分配，也支持使用 partitions 参数和 replication-factor 参数指定分区个数和副本个数进行自动分配。之后该方法会将副本分配结果写入到 ZooKeeper 中。
 

注意：Kafka 从 2.2 版本开始将 kafka-topic.sh 脚本中的 −−zookeeper 参数标注为 “过时”，推荐使用 −−bootstrap-server 参数。若读者依旧使用的是 2.1 及以下版本，请将下述的 --bootstrap-server 参数及其值手动替换为 --zookeeper zk1:2181,zk2:2181,zk:2181。一定要注意两者参数值所指向的集群地址是不同的。

形式一：

使用 replica-assignment 参数手动指定 Topic Partition Replica 与 Kafka Broker 之间的存储映射关系。

[root@localhost kafka]# kafka-topics.sh --create --bootstrap-server localhost:9092  --topic test6 --replica-assignment 1:0,1:0,1:0
 
 
[root@localhost kafka]# ls kafka-logs/test6-
test6-0/ test6-1/ test6-2/

• 格式：broker-id:副本数量,broker-id:副本数量,broker-id:副本数量
• 注意：broker-id是broker-server的id号；副本数量不能大于broker的数量（我这里就一台broker）；分区数量是用“,” 逗号分隔
•  注意：1:0,1:0,1:0 中的数字均为 broker.id；3个分区(逗号分隔)；每个分区有一个副本(副本所在的 broker 以冒号分割)。
• 此形式在最新的 2.3 版本中会报 Aborted due to timeout 异常，建议使用形式二。

形式二，使用 partitions 和 replication-factor 参数自动分配存储映射关系

1. 创建topic（4个分区，1个副本[前提是副本数量小于等于broker数量，我目前就一个broker]）

注意：Topic 名称中一定不要同时出现下划线 (’_’) 和小数点 (’.’)。
WARNING: Due to limitations in metric names, topics with a period (’.’) or underscore(’_’) could collide. To avoid issues ot os best to use either, but not both.

错误示范： 

[root@localhost kafka]# kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 4 --topic test
Error while executing topic command : Replication factor: 2 larger than available brokers: 1.
[2022-08-16 15:59:09,119] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 2 larger than available brokers: 1.
 (kafka.admin.TopicCommand$)

正确示范：

[root@localhost kafka]# kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 4 --topic test
Created topic test.
[root@localhost kafka]# ls kafka-logs/
cleaner-offset-checkpoint  log-start-offset-checkpoint  meta.properties  recovery-point-offset-checkpoint  replication-offset-checkpoint  test-0  test-1  test-2  test-3

kafka版本 >= 2.2

[root@localhost kafka]# kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test1
[root@localhost kafka]# ls kafka-logs/
cleaner-offset-checkpoint  log-start-offset-checkpoint  meta.properties  recovery-point-offset-checkpoint  replication-offset-checkpoint  test-0  test-1  test1-0  test-2  test-3

•  --create：指定创建topic动作
• --topic：指定新建topic的名称
• --zookeeper： 指定kafka连接zk的连接url，该值和server.properties文件中的配置项{zookeeper.connect}一样
• --partitions：指定当前创建的kafka分区数量，默认为1个
• --replication-factor：指定每个分区的复制因子个数，默认1个

创建 Topic 时也可指定参数：

[root@localhost kafka]#  kafka-topics.sh --create --bootstrap-server localhost:9092 --topic test8 --partitions 3 --replication-factor 1 --config cleanup.policy=compact --config retention.ms=500

2 # 查看所有topic的详细信息 

[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe
Topic:test	PartitionCount:4	ReplicationFactor:1	Configs:
	Topic: test	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 2	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 3	Leader: 1	Replicas: 1	Isr: 1
Topic:test1	PartitionCount:1	ReplicationFactor:1	Configs:
	Topic: test1	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
 
 
#查询指定topic的详细信息
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test
Topic:test	PartitionCount:4	ReplicationFactor:1	Configs:
	Topic: test	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 2	Leader: 1	Replicas: 1	Isr: 1
	Topic: test	Partition: 3	Leader: 1	Replicas: 1	Isr: 1
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test1
Topic:test1	PartitionCount:1	ReplicationFactor:1	Configs:
	Topic: test1	Partition: 0	Leader: 1	Replicas: 1	Isr: 1

•     PartitionCount：partition 个数。
•  ReplicationFactor：副本个数。
•  Partition：partition 编号，从 0 开始递增。
•     Leader：当前 partition 起作用的 breaker.id。
•    Replicas: 当前副本数据所在的 breaker.id，是一个列表，排在最前面的其作用。
•    Isr：当前 kakfa 集群中可用的 breaker.id 列表

3. 查询topic列表

[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
 
test
test1
 
#查询topic列表（支持0.9版本+）
 
[root@localhost kafka]# kafka-topics.sh --list --bootstrap-server localhost:9092
test
test1

  4. # 分区扩容，注意：分区数量只能增加，不能减少；不能用来修改副本个数。(请使用 kafka-reassign-partitions.sh 脚本增加副本数)

#kafka版本 < 2.2
[root@localhost kafka]# kafka-topics.sh --zookeeper localhost:2181 --alter --topic test1 --partitions 2
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test1
Topic:test1	PartitionCount:2	ReplicationFactor:1	Configs:
	Topic: test1	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: test1	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
 
 
# kafka版本 >= 2.2
[root@localhost kafka]# kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic test1 --partitions 3
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test1
Topic:test1	PartitionCount:3	ReplicationFactor:1	Configs:
	Topic: test1	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: test1	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: test1	Partition: 2	Leader: 1	Replicas: 1	Isr: 1

5.# 删除topic

若 delete.topic.enable=true
 直接彻底删除该 Topic。
若 delete.topic.enable=false
 如果当前 Topic 没有使用过即没有传输过信息：可以彻底删除。
 如果当前 Topic 有使用过即有过传输过信息：并没有真正删除 Topic 只是把这个 Topic 标记为删除(marked for deletion)，重启 Kafka Server 后删除。

注：delete.topic.enable=true 配置信息位于配置文件 config/server.properties 中(较新的版本中无显式配置，默认为 true)。

[root@localhost kafka]# kafka-topics.sh --zookeeper localhost:2181 --delete --topic test1
Topic test1 is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
 
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test1
Error while executing topic command : Topics in [] does not exist
[2022-08-16 16:08:55,601] ERROR java.lang.IllegalArgumentException: Topics in [] does not exist
	at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:416)
	at kafka.admin.TopicCommand$ZookeeperTopicService.describeTopic(TopicCommand.scala:332)
	at kafka.admin.TopicCommand$.main(TopicCommand.scala:66)
	at kafka.admin.TopicCommand.main(TopicCommand.scala)

6.增加配置

kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --config flush.messages=1

 7.删除配置

kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --delete-config flush.messages

Topic 级别配置属性 

当如下所示的属性配置到 Topic 上时，将会覆盖 server.properties 上对应的属性。

生产者相关

生产消息有存入到分区中，有三种方式。【1.指定分区；2.带有key；3.无key】；

kafka-console-producer.sh 脚本通过调用 kafka.tools.ConsoleProducer 类加载命令行参数的方式，在控制台生产消息的脚本。

三个分区一个副本的topic 

[root@localhost kafka]# kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic test1
 
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic test1
Topic:test1	PartitionCount:3	ReplicationFactor:1	Configs:
	Topic: test1	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: test1	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: test1	Partition: 2	Leader: 1	Replicas: 1	Isr: 1

1.无key型消息

默认情况下，所生产的消息是没有 key 的，命令如下：

执行上述命令后，就会在控制台等待键入消息体，直接输入消息值(value)即可，每行（以换行符分隔）表示一条消息，如下所示。

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic test1
>"Hello kafka"

    正常情况，每次回车表示触发“发送”操作，回车后可直接使用“Ctrl + c”退出生产者控制台，再使用  kafka-console-consumer.sh 脚本验证本次的生产情况。

2. 有key型消息

当需要为消息指定 key 时，可使用如下命令：

默认消息键与消息值间使用“Tab键”进行分隔，切勿使用转义字符(\t)，如下所示：

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic test1 --property parse.key=true
>name   kafka
>age	11

键入如上信息表示所生产的消息“name”为消息键，“kafka”为消息值。

3.指定分区

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic test1 1

附表 属性列表

 下表列举了当前版本支持的所有参数用法，敬请参阅。

消费者相关

--reset-offsets 后面需要接重置的模式1. 基本操作

#创建一个topic 一个副本 三个分区
[root@localhost kafka]# kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic jettech
 
 
# 生产者
[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic jettech
>"hello jettech"
 
 
#消费者,其中"--from-beginning"为可选参数，表示要从头消费消息    
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech --from-beginning
"hello jettech"
 
#指定groupid
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech --from-beginning --consumer-property group.id=jettoloader-consumer-group
"hello jettech"
 
#或指定文件
[root@localhost consume]# cat consume.properties 
group.id=consumer-group-jettoloader
[root@localhost consume]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech  --consumer.config  consume.properties 
"wubo"
 
 
# 指定分区发送数据
[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic jettech 2
>"hello beijing"
>"nihao"
# 指定分区接受数据
[root@localhost consume]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech --from-beginning --partition 1
>"hello beijing"
>"nihao"
 
 
# 新生产者（支持0.9版本+）
[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092 --topic beijing --producer.config config/producer.properties
>"hello bejing"
 
 
# 新消费者（支持0.9版本+）
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic beijing  --from-beginning --consumer.config config/consumer.properties
"hello bejing"
 
 
# kafka-verifiable-consumer.sh（消费者事件，例如：offset提交等
[root@localhost kafka]# kafka-verifiable-consumer.sh --broker-list localhost:9092 --topic jettech --group-id consumer-group-jettoloader
{"timestamp":1660702053129,"name":"startup_complete"}
{"timestamp":1660702053323,"name":"partitions_revoked","partitions":[]}
{"timestamp":1660702053344,"name":"partitions_assigned","partitions":[{"topic":"jettech","partition":0},{"topic":"jettech","partition":2},{"topic":"jettech","partition":1}]}
{"timestamp":1660702053395,"name":"records_consumed","count":3,"partitions":[{"topic":"jettech","partition":0,"count":1,"minOffset":0,"maxOffset":0},{"topic":"jettech","partition":2,"count":1,"minOffset":0,"maxOffset":0},{"topic":"jettech","partition":1,"count":1,"minOffset":2,"maxOffset":2}]}
{"timestamp":1660702053403,"name":"offsets_committed","offsets":[{"topic":"jettech","partition":0,"offset":1},{"topic":"jettech","partition":2,"offset":1},{"topic":"jettech","partition":1,"offset":3}],"success":true}
 
 
#查看消费者组
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
consumer-group-jettoloader
test-consumer-group
jettoloader-consumer-group
 
#查看消费者组详情--describe
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group jettoloader-consumer-group
Consumer group 'jettoloader-consumer-group' has no active members.
 
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
jettech         0          0               1               1               -               -               -
jettech         2          0               1               1               -               -               -
jettech         1          1               3               2               -               -               -
 
#查询消费者成员信息--members
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group jettech-hello1-comsume-group --members
 
CONSUMER-ID                                     HOST            CLIENT-ID       #PARTITIONS     
consumer-1-2d7d59c3-64c9-4526-905b-4e8920db3e84 /192.168.1.36   consumer-1      2 
 
#查询消费者状态信息--state
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group jettech-hello1-comsume-group --state
 
COORDINATOR (ID)          ASSIGNMENT-STRATEGY       STATE                #MEMBERS
192.168.1.36:9092 (1)     range                     Stable               1
 
 
 
#删除消费者组--delete
 
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group jettech-hellp-comsume-group01
Deletion of requested consumer groups ('jettech-hellp-comsume-group01') was successful.
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
jettech-hellp-comsume-group
consumer-group-jettoloader
jettech-hello1-comsume-group
test-consumer-group
jettoloader-consumer-group
 
#PS: 想要删除消费组前提是这个消费组的所有客户端都停止消费/不在线才能够成功删除;否则会报下面异常
[root@localhost kafka]# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group jettech-hello1-comsume-group
Error: Deletion of some consumer groups failed:
* Group 'jettech-hello1-comsume-group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.
 
 
#重置消费组的偏移量 --reset-offsets
#能够执行成功的一个前提是 消费组这会是不可用状态;
下面的示例使用的参数是: --dry-run ;这个参数表示预执行,会打印出来将要处理的结果;
等你想真正执行的时候请换成参数--excute ;
下面示例 重置模式都是 --to-earliest 重置到最早的;
请根据需要参考下面 相关重置Offset的模式 换成其他模式;
重置指定消费组的偏移量 --group
 
 
#重置指定消费组的指定Topic的偏移量到最晚offset  【--to-latest 】
[root@localhost kafka]# kafka-consumer-groups.sh --reset-offsets --to-earliest --group jettech-hellp1-comsume-group --bootstrap-server localhost:9092 --dry-run --topic hello1
 
TOPIC                          PARTITION  NEW-OFFSET     
hello1                         0          0              
hello1                         1          0
 
 
 
#重置指定消费组的指定Topic的偏移量到指定的offset  【--to-latest 】
1)获取offset
[root@localhost kafka]# kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic jettech --time -1
jettech:0:33334
jettech:1:33336
jettech:2:33335
 
2）重置指定消费组的指定Topic的偏移量到指定的offset
[root@localhost kafka]# kafka-consumer-groups.sh --reset-offsets  --to-offset 33336 --group jettech-hellp1-comsume-group --bootstrap-server localhost:9092 --dry-run --topic hello1
 
TOPIC                          PARTITION  NEW-OFFSET     
hello1                         0          0              
hello1                         1          0

--reset-offsets 后面需要接重置的模式:相关重置Offset的模式

--to-earliest :    重置offset到最开始的那条offset(找到还未被删除最早的那个offset)    
--to-current:    直接重置offset到当前的offset，也就是LOE    
--to-latest：    重置到最后一个offset    
--to-datetime:    重置到指定时间的offset;格式为:YYYY-MM-DDTHH:mm:SS.sss;    --to-datetime "2021-6-26T00:00:00.000"
--to-offset    重置到指定的offset,但是通常情况下,匹配到多个分区,这里是将匹配到的所有分区都重置到这一个值; 如果 1.目标最大offset<--to-offset, 这个时候重置为目标最大offset；2.目标最小offset>--to-offset ，则重置为最小; 3.否则的话才会重置为--to-offset的目标值; 一般不用这个    --to-offset 3465 
--shift-by    按照偏移量增加或者减少多少个offset；正的为往前增加;负的往后退；当然这里也是匹配所有的;    --shift-by 100 、--shift-by -100
--from-file    根据CVS文档来重置; 这里下面单独讲解    

上面其他的一些模式重置的都是匹配到的所有分区; 不能够每个分区重置到不同的offset；不过**--from-file**可以让我们更灵活一点;

1先配置cvs文档
格式为: Topic:分区号: 重置目标偏移量

test2,0,100
test2,1,200
test2,2,300

2执行命令

kafka-consumer-groups.sh --reset-offsets --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --from-file config/reset-offset.csv

通用命令kafka-run-class.sh

运行一个class，调用kafka的tools的部分功能。

kafka-run-class.sh [-daemon] [-name servicename] [-loggc] classname [opts]

GetOffsetShell

kafka-run-class.sh kafka.tools.GetOffsetShell /?
 
 
 
#获取offset
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list kafka1:9092 --topic jettech --time -1

options：
Option    Description
–broker-list    
–max-wait-ms <Integer: ms>    废弃。(default: 1000)
–offsets <Integer: count>    废弃。(default: 1)
–partitions <String: partition ids>    partition id列表
–time <Long: >    时间戳。返回指定时间戳之前的offset。 timestamp/-1(latest，默认值)/-2(earliest)。如果时间戳大于当前时刻，无offset返回。
–topic <String: topic>    
 

kafka常用命令及问题解决_demon7552003的博客-CSDN博客

 配置文件：

1producer.properties:生产端的配置文件

#指定kafka节点列表，用于获取metadata，不必全部指定
#需要kafka的服务器地址，来获取每一个topic的分片数等元数据信息。
metadata.broker.list=kafka01:9092,kafka02:9092,kafka03:9092
 
#生产者生产的消息被发送到哪个block，需要一个分组策略。
#指定分区处理类。默认kafka.producer.DefaultPartitioner，表通过key哈希到对应分区
#partitioner.class=kafka.producer.DefaultPartitioner
 
#生产者生产的消息可以通过一定的压缩策略（或者说压缩算法）来压缩。消息被压缩后发送到broker集群，
#而broker集群是不会进行解压缩的，broker集群只会把消息发送到消费者集群，然后由消费者来解压缩。
#是否压缩，默认0表示不压缩，1表示用gzip压缩，2表示用snappy压缩。
#压缩后消息中会有头来指明消息压缩类型，故在消费者端消息解压是透明的无需指定。
#文本数据会以1比10或者更高的压缩比进行压缩。
compression.codec=none
 
#指定序列化处理类，消息在网络上传输就需要序列化，它有String、数组等许多种实现。
serializer.class=kafka.serializer.DefaultEncoder
 
#如果要压缩消息，这里指定哪些topic要压缩消息，默认empty，表示不压缩。
#如果上面启用了压缩，那么这里就需要设置
#compressed.topics= 
#这是消息的确认机制，默认值是0。在面试中常被问到。
#producer有个ack参数，有三个值，分别代表：
#（1）不在乎是否写入成功；
#（2）写入leader成功；
#（3）写入leader和所有副本都成功；
#要求非常可靠的话可以牺牲性能设置成最后一种。
#为了保证消息不丢失，至少要设置为1，也就
#是说至少保证leader将消息保存成功。
#设置发送数据是否需要服务端的反馈,有三个值0,1,-1，分别代表3种状态：
#0: producer不会等待broker发送ack。生产者只要把消息发送给broker之后，就认为发送成功了，这是第1种情况；
#1: 当leader接收到消息之后发送ack。生产者把消息发送到broker之后，并且消息被写入到本地文件，才认为发送成功，这是第二种情况；#-1: 当所有的follower都同步消息成功后发送ack。不仅是主的分区将消息保存成功了，
#而且其所有的分区的副本数也都同步好了，才会被认为发动成功，这是第3种情况。
request.required.acks=0
 
#broker必须在该时间范围之内给出反馈，否则失败。
#在向producer发送ack之前,broker允许等待的最大时间 ，如果超时,
#broker将会向producer发送一个error ACK.意味着上一次消息因为某种原因
#未能成功(比如follower未能同步成功)
request.timeout.ms=10000
 
#生产者将消息发送到broker，有两种方式，一种是同步，表示生产者发送一条，broker就接收一条；
#还有一种是异步，表示生产者积累到一批的消息，装到一个池子里面缓存起来，再发送给broker，
#这个池子不会无限缓存消息，在下面，它分别有一个时间限制（时间阈值）和一个数量限制（数量阈值）的参数供我们来设置。
#一般我们会选择异步。
#同步还是异步发送消息，默认“sync”表同步，"async"表异步。异步可以提高发送吞吐量,
#也意味着消息将会在本地buffer中,并适时批量发送，但是也可能导致丢失未发送过去的消息
producer.type=sync
 
#在async模式下,当message被缓存的时间超过此值后,将会批量发送给broker,
#默认为5000ms
#此值和batch.num.messages协同工作.
queue.buffering.max.ms = 5000
 
#异步情况下，缓存中允许存放消息数量的大小。
#在async模式下,producer端允许buffer的最大消息量
#无论如何,producer都无法尽快的将消息发送给broker,从而导致消息在producer端大量沉积
#此时,如果消息的条数达到阀值,将会导致producer端阻塞或者消息被抛弃，默认为10000条消息。
queue.buffering.max.messages=20000
 
#如果是异步，指定每次批量发送数据量，默认为200
batch.num.messages=500
 
#在生产端的缓冲池中，消息发送出去之后，在没有收到确认之前，该缓冲池中的消息是不能被删除的，
#但是生产者一直在生产消息，这个时候缓冲池可能会被撑爆，所以这就需要有一个处理的策略。
#有两种处理方式，一种是让生产者先别生产那么快，阻塞一下，等会再生产；另一种是将缓冲池中的消息清空。
#当消息在producer端沉积的条数达到"queue.buffering.max.meesages"后阻塞一定时间后,
#队列仍然没有enqueue(producer仍然没有发送出任何消息)
#此时producer可以继续阻塞或者将消息抛弃,此timeout值用于控制"阻塞"的时间
#-1: 不限制阻塞超时时间，让produce一直阻塞,这个时候消息就不会被抛弃
#0: 立即清空队列,消息被抛弃
queue.enqueue.timeout.ms=-1
 
 
#当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数
#因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失)
#有可能导致broker接收到重复的消息,默认值为3.
message.send.max.retries=3
 
#producer刷新topic metada的时间间隔,producer需要知道partition leader
#的位置,以及当前topic的情况
#因此producer需要一个机制来获取最新的metadata,当producer遇到特定错误时,
#将会立即刷新
#(比如topic失效,partition丢失,leader失效等),此外也可以通过此参数来配置
#额外的刷新机制，默认值600000
topic.metadata.refresh.interval.ms=60000

2).consumer.properties:消费端的配置文件

#消费者集群通过连接Zookeeper来找到broker。
#zookeeper连接服务器地址
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181
 
#zookeeper的session过期时间，默认5000ms，用于检测消费者是否挂掉
zookeeper.session.timeout.ms=5000
 
#当消费者挂掉，其他消费者要等该指定时间才能检查到并且触发重新负载均衡
zookeeper.connection.timeout.ms=10000
 
#这是一个时间阈值。
#指定多久消费者更新offset到zookeeper中。
#注意offset更新时基于time而不是每次获得的消息。
#一旦在更新zookeeper发生异常并重启，将可能拿到已拿到过的消息
zookeeper.sync.time.ms=2000
 
#指定消费
group.id=xxxxx
 
#这是一个数量阈值，经测试是500条。
#当consumer消费一定量的消息之后,将会自动向zookeeper提交offset信息#注意offset信息并不是每消费一次消息就向zk提交
#一次,而是现在本地保存(内存),并定期提交,默认为true
auto.commit.enable=true
 
# 自动更新时间。默认60 * 1000
auto.commit.interval.ms=1000
 
# 当前consumer的标识,可以设定,也可以有系统生成,
#主要用来跟踪消息消费情况,便于观察
conusmer.id=xxx
 
# 消费者客户端编号，用于区分不同客户端，默认客户端程序自动产生
client.id=xxxx
 
# 最大取多少块缓存到消费者(默认10)
queued.max.message.chunks=50
 
# 当有新的consumer加入到group时,将会reblance,此后将会
#有partitions的消费端迁移到新  的consumer上,如果一个
#consumer获得了某个partition的消费权限,那么它将会向zk
#注册 "Partition Owner registry"节点信息,但是有可能
#此时旧的consumer尚没有释放此节点, 此值用于控制,
#注册节点的重试次数.
rebalance.max.retries=5
 
#每拉取一批消息的最大字节数
#获取消息的最大尺寸,broker不会像consumer输出大于
#此值的消息chunk 每次feth将得到多条消息,此值为总大小,
#提升此值,将会消耗更多的consumer端内存
fetch.min.bytes=6553600
 
#当消息的尺寸不足时,server阻塞的时间,如果超时,
#消息将立即发送给consumer
#数据一批一批到达，如果每一批是10条消息，如果某一批还
#不到10条，但是超时了，也会立即发送给consumer。
fetch.wait.max.ms=5000
socket.receive.buffer.bytes=655360
 
# 如果zookeeper没有offset值或offset值超出范围。
#那么就给个初始的offset。有smallest、largest、
#anything可选，分别表示给当前最小的offset、
#当前最大的offset、抛异常。默认largest
auto.offset.reset=smallest
 
# 指定序列化处理类
derializer.class=kafka.serializer.DefaultDecoder

3).server.properties:服务端的配置文件

#broker的全局唯一编号，不能重复
broker.id=0
 
#用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092
 
#处理网络请求的线程数量，也就是接收消息的线程数。
#接收线程会将接收到的消息放到内存中，然后再从内存中写入磁盘。
num.network.threads=3
 
#消息从内存中写入磁盘是时候使用的线程数量。
#用来处理磁盘IO的线程数量
num.io.threads=8
 
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
 
#接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400
 
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
 
#kafka运行日志存放的路径
log.dirs=/export/servers/logs/kafka
 
#topic在当前broker上的分片个数
num.partitions=2
 
#我们知道segment文件默认会被保留7天的时间，超时的话就
#会被清理，那么清理这件事情就需要有一些线程来做。这里就是
#用来设置恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
 
#segment文件保留的最长时间，默认保留7天（168小时），
#超时将被删除，也就是说7天之前的数据将被清理掉。
log.retention.hours=168
 
#滚动生成新的segment文件的最大时间
log.roll.hours=168
 
#日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824
 
#上面的参数设置了每一个segment文件的大小是1G，那么
#就需要有一个东西去定期检查segment文件有没有达到1G，
#多长时间去检查一次，就需要设置一个周期性检查文件大小
#的时间（单位是毫秒）。
log.retention.check.interval.ms=300000
 
#日志清理是否打开
log.cleaner.enable=true
 
#broker需要使用zookeeper保存meta数据
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181
 
#zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000
 
#上面我们说过接收线程会将接收到的消息放到内存中，然后再从内存
#写到磁盘上，那么什么时候将消息从内存中写入磁盘，就有一个
#时间限制（时间阈值）和一个数量限制（数量阈值），这里设置的是
#数量阈值，下一个参数设置的则是时间阈值。
#partion buffer中，消息的条数达到阈值，将触发flush到磁盘。
log.flush.interval.messages=10000
 
#消息buffer的时间，达到阈值，将触发将消息从内存flush到磁盘，
#单位是毫秒。
log.flush.interval.ms=3000
 
#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true
 
#此处的host.name为本机IP(重要),如果不改,则客户端会抛出:
#Producer connection to localhost:9092 unsuccessful 错误!
host.name=kafka01
 
advertised.host.name=192.168.239.128

日志文件的删除策略非常简单:启动一个后台线程定期扫描log file列表,把保存时间超过阀值的文件直接删除(根据文件的创建时间).清理参数在server.properties文件中

Kafka日志管理器允许定制删除策略。目前的策略是删除修改时间在N天之前的日志（按时间删除），也可以使用另外一个策略：保留最后的N GB数据的策略(按大小删除)。为了避免在删除时阻塞读操作，采用了copy-on-write形式的实现，删除操作进行时，读取操作的二分查找功能实际是在一个静态的快照副本上进行的，这类似于Java的CopyOnWriteArrayList。

Kafka消费日志删除思想：Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用

Kafka日志管理器允许压缩策略

将数据压缩，只保留每个key最后一个版本的数据。首先在broker的配置中设置log.cleaner.enable=true启用cleaner，这个默认是关闭的。在Topic的配置中设置log.cleanup.policy=compact启用压缩策略。

 在整个数据流中，每个Key都有可能出现多次，压缩时将根据Key将消息聚合，只保留最后一次出现时的数据。这样，无论什么时候消费消息，都能拿到每个Key的最新版本的数据。
    压缩后的offset可能是不连续的，比如上图中没有5和7，因为这些offset的消息被merge了，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，比如，当试图获取offset为5的消息时，实际上会拿到offset为6的消息，并从这个位置开始消费。
    这种策略只适合特俗场景，比如消息的key是用户ID，消息体是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。
    压缩策略支持删除，当某个Key的最新版本的消息没有内容时，这个Key将被删除，这也符合以上逻辑。

• 消息消费

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicName
  

  表示从 latest 位移位置开始消费该主题的所有分区消息，即仅消费正在写入的消息。

• 从开始位置消费

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic topicName
  

   表示从指定主题中有效的起始位移位置开始消费所有分区的消息

• 显示key消费

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --property print.key=true --topic topicName
  

  消费出的消息结果将打印出消息体的 key 和 value。

• 执行分区消费

  # 指定分区接受数据
  [root@localhost consume]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech --from-beginning --partition 1
  

  若还需要为你的消息添加其他属性，请参考下述列表。

kafka发送、消费指定分区消息_月亮船长的博客-CSDN博客_kafka消费指定分区

kafka-consumer-groups.sh消费者组管理【资料】

什么是kafka消费者组
  kafka消费者组（Consumer Group）是kafka提供的可扩展且具有容错性的消费者机制。
  它是一个组，所以内部有可以有多个消费者，这些消费者共用一个ID（Group ID），一个组内的所有消费者共同协作，完成对订阅的主题的所有分区进行消费。其中一个主题中的一个分区只能由一个消费者消费。

消费者组的特性

1. 一个消费者组可以有多个消费者。
2. Group ID是一个字符串，在一个kafka集群中，它标识唯一的一个消费者组。
3. 每个消费者组订阅的所有主题中，每个主题的每个分区只能由一个消费者消费。消费者组之间不影响。

为什么出现消费者组

        我们知道的消息引擎模型有：点对点模型和发布/订阅模型。传统的消息引擎就是这两大类。这两大类消息引擎，都有各自适合的应用场景，也都有不适应的场景。
  点对点的模型，每消费一个消息之后，被消费的消息就会被删除。如果我们需要多个消费者消费同一个消息队列时，就不能使用点对点模型了。
  发布订阅模型，支持多个消费者消费同一个消息队列，但是发布订阅模型中，消费者订阅了一个主题后，就要订阅主题的所有分区。这总方式既不灵活，也会影响消息的真是投递效果。
  消费者组就避开了上述两种模型的缺陷，有兼容了他们的优点。
  首先消费者之间彼此独立，互不影响。可以订阅同一个主题并且互不干扰。再加上Broker端的消息留存机制，kafka的消费者组就完美的解决了上面的问题。kafka使用一种消费者组（Consumer Group）机制，就同时实现了传统消息引擎系统的两大模型：如果所有的消费者实例都属于一个消费者组那就是点对点模型，如果所有消费者实例各自是独立的消费者那就是发布订阅模型。
  因为上面消费组的第三个特性。所以消费者组的消费者实例数最好等于该消费者组订阅的主题中的分区数。如果实例数量多于分区数，那多余的实例将永远不会工作，除非有其他实例挂掉。

针对Consumer Group,Kafka如何管理位移（offset）？

        这个问题需要区分新老版本。首先他们的存储方式都是使用类似于map的KV对实现的存储。key是分区，v对应Consumer消费该分区的最新位移。我们可以这样理解，但是实际的存储要比这个复杂的多。
  新老版本的区别在于位移存储的位置：老版本是将位移存放到zookeeper中，而zookeeper是一个分布式的协调服务框架，kafka重度依赖它实现的各种各样的协调管理。将位移存到zookeeper中的做法，显而易见的好处是减少了kafka broker端的状态保存开销。可以自由的扩缩容，实现超强的伸缩性。
  但是由于zookeeper这类元框架其实并不适合进行频繁的写更新。而Consumer Group的位移更新却是一个非常频繁的更新操作。所以并不是很适合将位移存在zookeeper中。
  于是新版本将位移保存在kafka内部主题中。就是：_consumer_offsets。

kafka的Rebalance(重平衡)

定义
  Rebalance本质上是一种协议，规定了一个Consumer Group下的所有Consumer如何达成一致，来分配订阅Topic的每个分区。例如：某个Group下有20个Consumer实例，它订阅了一个有100个分区的Topic。正常情况下，kafka会给每个实例分配5个分区。这个分配的过程叫做Rebalance。
  rebalance发生时，Group下的所有Consumer实例都会协调在一起共同参与。具体怎么分配，是有分配策略协助的。分配策略以后再总结。

触发条件

1. 组成员数发生变化。比如有实例进入或者离开组。
2. 订阅的主题数发生变更。
3. 订阅主题的分区数发生变更。

问题

  rebalance有一个比较大的问题。那就是再Rebalance过程中，所有的实例都会停止消费，等Rebalance完成。这就导致Rebalance过程中无法提供服务。而且，Rebalance的过程还很慢。所以我们要尽量避免Rebalance的发生。

案例操作：

案例一：消费者多于分区数

1）创建生产者1个分区，创建生产者一个副本 

[root@localhost kafka]# kafka-topics.sh --zookeeper localhost:2181 --create --topic hello --partitions 1 --replication-factor 1
Created topic hello.
 
[root@localhost kafka]# kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets
beijing
hello
jettech
shanghai
test
test1
test1-1
test2
test3
test4
test5
test6
test8
topicname
[root@localhost kafka]# kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic hello
Topic:hello	PartitionCount:1	ReplicationFactor:1	Configs:
	Topic: hello	Partition: 0	Leader: 1	Replicas: 1	Isr: 1

2）启动生产者

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092  --topic hello
>"hello hello:"

3）创建两个消费者

[root@localhost kafka]# cat  consume/consume.properties 
group.id=jettech-hellp-comsume-group

两个消费者都是用同一台机器,开两个shell终端操作的.
下面命令在两个shell终端分别输入一次

[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume.properties 
wubo
 
 
 
 
[root@localhost kafka]#  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume.properties

生产者生产数据

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092  --topic hello
>"hello hello:"
>wubo
>aaa
>bbb
>cccc
>dddd
>eeee
>fff

消费者1接收数据

[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume.properties 
wubo
aaa
bbb
cccc
dddd
eeee
fff

消费者2接收数据

[root@localhost kafka]#  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume.properties

结果:

消费者1接收到了7条数据, 消费者2一条数据没接收到.

结论 同一个消费者组内同一时刻只有一个消费者接收到消息。 另外一个消费者分不到partition消费就空着待机了

因为启动consumer的时候,consumer会通过某种算法分配消费哪个partition, 分配完了之后这个consumer就一直消费这个partition的数据了.别的分区这个consumer就管都不会管了.

案例二：不同消费者组消费相同topic的相同分区数据

不同消费者组的消费者是可以同时消费同一个topic的同一个分区信息的

1）生产：

 
[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092  --topic hello
>aaa
>bbbb
>ccc
>ddd
>

2）消费1

[root@localhost kafka]# cat consume/consume.properties 
group.id=jettech-hellp-comsume-group
 
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume.properties 
aaaaa
bbbb
cccc

3）消费2

[root@localhost kafka]# cat consume/consume1.properties 
group.id=jettech-hellp-comsume-group1
 
[root@localhost kafka]#  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello  --consumer.config  consume/consume1.properties 
aaaaa
bbbb
cccc

结果:

消费者1接收到了3条数据, 消费者2也接收到了3条数据

结论 不同消费者组的消费者可同时消费同一个topic的同一个分区。

案例三：生产者2个分区

1）重新创建一个生产者,是2个分区的.,

[root@localhost kafka]# kafka-topics.sh --zookeeper localhost:2181 --create --topic hello1 --partitions 2 --replication-factor 1
Created topic hello1.

2）创建两个消费者

两个消费者都是用相同机器,开两个shell终端操作的.
下面命令在两个shell终端分别输入一次

[root@localhost kafka]# kafka-console-producer.sh --broker-list localhost:9092  --topic hello1
>hello1
>hello2
>hello3
>hello4

3）消费1

[root@localhost kafka]# cat consume/consume.properties 
group.id=jettech-hello1-comsume-group
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello1  --consumer.config  consume/consume.properties 
hello1
hello3

4）消费2

[root@localhost kafka]# cat consume/consume.properties 
group.id=jettech-hello1-comsume-group
[root@localhost kafka]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello1  --consumer.config  consume/consume.properties 
hello2
hello4

结论: 消费者接收数据被轮询了

Kafka中的消费者组（Consumer Group） - DoubleLi - 博客园

切换leader

# kafka版本 <= 2.4
kafka-preferred-replica-election.sh --zookeeper zk_host:port/chroot
 
# kafka新版本
kafka-preferred-replica-election.sh --bootstrap-server broker_host:port

kafka自带压测命令

[root@localhost kafka]# kafka-producer-perf-test.sh --topic shanghai --num-records 100 --record-size 1 --throughput 100  --producer-props bootstrap.servers=localhost:9092
[2022-08-17 10:13:57,158] WARN [Producer clientId=producer-1] Error while fetching metadata with correlation id 1 : {shanghai=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
100 records sent, 99.403579 records/sec (0.00 MB/sec), 8.25 ms avg latency, 299.00 ms max latency, 2 ms 50th, 16 ms 95th, 299 ms 99th, 299 ms 99.9th.

kafka持续发送消息

持续发送消息到指定的topic中，且每条发送的消息都会有响应信息：

#生产
[root@localhost kafka]# kafka-verifiable-producer.sh --broker-list $(hostname -f):9092 --topic jettech --max-messages 100000
 
#消费
[root@localhost kafka]#  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jettech --from-beginning

zookeeper-shell.sh

如果kafka集群的zk配置了chroot路径，那么需要加上/path。

 
[root@localhost kafka]# zookeeper-shell.sh localhost:2181[/path]
ls /brokers/ids
get /brokers/ids/0

迁移分区

1. 创建规则json

   kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file increase-replication-factor.json --execute
   

2. 验证

   kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file increase-replication-factor.json --verify
   

MirrorMaker 跨机房灾备工具

kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB