启动Kafka并生产消费消息
启动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
生产者发送消息
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并转发到另一个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参数
bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
2、查看kafka特定topic的详情，使用--topic与--describe参数
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、修改分区数
bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MY-TOPIC --partitions 10
5、查看topic某分区偏移量最大（小）值
./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参数
查看consumer group列表有新、旧两种命令，分别查看新版consumer列表（信息保存在broker中）和老版consumer列表（信息保存在zookeeper中），因而需要区分指定bootstrap--server和zookeeper参数：
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参数
同样根据新/旧版本的consumer，分别指定bootstrap-server与zookeeper参数:
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执行输出如下：
[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，发现并没有改变。
[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
参考：
