Kafka常用命令

创建topic

$ kafka-topics.sh --bootstrap-server 192.168.32.187:9092 --create --partitions 3 --replication-factor 2 --topic test01
Created topic test01.

查看主题

$ kafka-topics.sh --bootstrap-server 192.168.32.187:9092 --describe --topic test01 
Topic: test01     TopicId: SPJw_NPhQk6nAxhs_ZmNLw PartitionCount: 3       ReplicationFactor: 2    Configs: segment.bytes=1073741824
        Topic: test01     Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
        Topic: test01     Partition: 1    Leader: 1       Replicas: 1,0   Isr: 1,0
        Topic: test01     Partition: 2    Leader: 0       Replicas: 0,2   Isr: 0,2

删除topic

$ kafka-topics.sh --bootstrap-server 192.168.32.187:9092 --delete --topic test01

查看partitions偏移量

# 测试生产者
$ kafka-console-producer.sh --bootstrap-server 192.168.32.187:9092 --topic test01

# 测试消费者
$ kafka-console-consumer.sh --bootstrap-server 192.168.32.187:9092 --topic test01 --from-beginning

# 查看partitions偏移量
$ kafka-get-offsets.sh --bootstrap-server 192.168.32.187:9092 --topic test01
test01:0:12
test01:1:8
test01:2:8

说明：

• topic名称:partitionID:偏移量
• 如果上面的测试生产者没有提前创建topic，会默认创建一个分区的topic

查看消费组的偏移量

生产者自带客户端

$ kafka-console-producer.sh --bootstrap-server 192.168.32.187:9092 --topic test02

消费者客户端，消费test02的topic数据。每消费一条数据休眠3秒，模拟数据积压。消费组id为consumer01

$ sudo pip install python-kafka
$ cat > consumerClient.py <<-EOF
from kafka import KafkaConsumer
import time

consumer = KafkaConsumer('test02', group_id='consumer01', bootstrap_servers=['192.168.32.187:9092'])
for msg in consumer:
    recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)
    print recv
    time.sleep(3)
EOF

查看消费组的消费偏移量

$ kafka-consumer-groups.sh --bootstrap-server 192.168.32.187:9092 --describe --group consumer01

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                             HOST            CLIENT-ID
consumer01      test02          0          46              51              5               kafka-python-2.0.2-1230d186-c2ca-4849-b71b-b118b3873a5c /192.168.32.189 kafka-python-2.0.2
consumer01      test02          1          63              66              3               kafka-python-2.0.2-1230d186-c2ca-4849-b71b-b118b3873a5c /192.168.32.189 kafka-python-2.0.2
consumer01      test02          2          44              50              6               kafka-python-2.0.2-1230d186-c2ca-4849-b71b-b118b3873a5c /192.168.32.189 kafka-python-2.0.2
consumer01      test02          3          58              59              1               kafka-python-2.0.2-1230d186-c2ca-4849-b71b-b118b3873a5c /192.168.32.189 kafka-python-2.0.2

LOG-END-OFFSET(总共偏移量) = CURRENT-OFFSET(当前消费的偏移量) + LAG(积压的数据)

删除消费者组

$ kafka-consumer-groups.sh --bootstrap-server 192.168.32.187:9092 --delete --group consumer01
Deletion of requested consumer groups ('consumer01') was successful.