啟動Kafka并生產(chǎn)消費消息
啟動ZooKeeper
# 啟動ZooKeeper
$ bin/zookeeper-server-start.sh config/zookeeper.properties
啟動Kafka
# 前臺啟動Kafka
$ bin/kafka-server-start.sh config/server.properties
# 后臺啟動方法1
$ nohup bin/kafka-server-start.sh config/server.properties 1>/dev/null 2>&1 &
# 后臺啟動方法2(推薦)
$ bin/kafka-server-start.sh -daemon config/server.properties
查看啟動后kafka的版本
ps -ef | grep kafka
生產(chǎn)者發(fā)送消息
bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic MY-TOPIC
消費消息時指定group id(0.11版本)
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic MY-TOPIC --consumer-property group.id=my_test_group
查看某個offset開始的消息摹迷,確定消費時需要跳過的消息
bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic MY-TOPIC --partition 0 --offset 110823 --max-messages 1
從歷史消息中過濾出指定的消息
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic MY-TOPIC --from-beginning | grep "1738124945-1623549602774"
消費某個topic并轉發(fā)到另一個topic
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic MY-TOPIC --offset 110823 --consumer-property group.id=replayworker \
| tee message.txt \
| ./kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic MY-TOPIC2
查看修改topic收擦、group命令
1栋猖、查看kafka topic列表,使用--list參數(shù)
bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
2、查看kafka特定topic的詳情望众,使用--topic與--describe參數(shù)
bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic MY-TOPIC --describe
3、查看失效副本
[root@kafka-192-168-3-149 kafka_2.11-0.11.0.3]# bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic MY-TOPIC --under-replicated-partitions
Topic: MY-TOPIC Partition: 0 Leader: 4 Replicas: 4,2,3 Isr: 4,3
Topic: MY-TOPIC Partition: 1 Leader: 3 Replicas: 2,3,4 Isr: 3,4
Topic: MY-TOPIC Partition: 2 Leader: 3 Replicas: 3,4,2 Isr: 4,3
Topic: MY-TOPIC Partition: 3 Leader: 4 Replicas: 4,3,2 Isr: 4,3
Topic: MY-TOPIC Partition: 4 Leader: 4 Replicas: 2,4,3 Isr: 4,3
Topic: MY-TOPIC Partition: 5 Leader: 3 Replicas: 3,2,4 Isr: 4,3
Topic: MY-TOPIC Partition: 6 Leader: 4 Replicas: 4,2,3 Isr: 4,3
Topic: MY-TOPIC Partition: 7 Leader: 3 Replicas: 2,3,4 Isr: 4,3
Topic: MY-TOPIC Partition: 8 Leader: 3 Replicas: 3,4,2 Isr: 4,3
4、修改分區(qū)數(shù)
bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MY-TOPIC --partitions 10
5虹统、查看topic某分區(qū)偏移量最大(小)值
./kafka-run-class.sh kafka.tools.GetOffsetShell --topic MY-TOPIC --time -1 --broker-list 127.0.0.1:9092 --partitions 0
6隧甚、查看consumer group列表车荔,使用--list參數(shù)
查看consumer group列表有新、舊兩種命令戚扳,分別查看新版consumer列表(信息保存在broker中)和老版consumer列表(信息保存在zookeeper中)忧便,因而需要區(qū)分指定bootstrap--server和zookeeper參數(shù):
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --list #新
bin/kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181 --list #舊
7、查看特定consumer group詳情(包含group的offset)帽借,使用--group與--describe參數(shù)
同樣根據(jù)新/舊版本的consumer珠增,分別指定bootstrap-server與zookeeper參數(shù):
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --group my_test_group --describe #新
bin/kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181 --group my_test_group --describe #舊
新版consumer執(zhí)行輸出如下:
[root@dami-03 bin]# ./kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --group my_test_group --describe
Note: This will only show information about consumers that use the Java consumer API (non-ZooKeeper-based consumers).
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
MY-TOPIC 0 73423 73423 0 rdkafka-52aa7145-45f6-403f-ac52-5aa3a69a85f1 /127.0.0.1 rdkafka
MY-TOPIC 1 73422 73422 0 rdkafka-52aa7145-45f6-403f-ac52-5aa3a69a85f1 /127.0.0.1 rdkafka
MY-TOPIC 2 73424 73424 0 rdkafka-52aa7145-45f6-403f-ac52-5aa3a69a85f1 /127.0.0.1 rdkafka
8、修改某個消費組的偏移量(offset)
0.11版本以下方法不生效
bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group my_test_group --topic MY-TOPIC:0 --reset-offsets --to-offset 113115 -–execute
輸出如下砍艾,查看offset蒂教,發(fā)現(xiàn)并沒有改變。
[root@local-01 kafka_2.11-0.11.0.3]# bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --group my_test_group --topic MY-TOPIC:0 --reset-offsets --to-offset 113115 –-execute
Note: This will only show information about consumers that use the Java consumer API (non-ZooKeeper-based consumers).
TOPIC PARTITION NEW-OFFSET
MY-TOPIC 0 113115
用這個方法
# offset值為 目標offset(113115) - 500 = 112615
bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic MY-TOPIC --partition 2 --offset 112615 --max-messages 1 --consumer-property group.id=my_test_group
參考: