本文操作基于 Docker comopose构建的本地kafka集群
1. 3个broker,创建一个topic(3个replication,4个partition)
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-topics.sh --create --topic userMsg --partitions 4 --zookeeper 172.19.0.11:2181 --replication-factor 3
Created topic userMsg.
开启一个生产者
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-console-producer.sh --bootstrap-server 0.0.0.0:9092 --topic userMsg
>msg1
2. 开启一个消费者,指定消费者组为myConsumerGroup. 生产者生产一条信息,消费者接收到。
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-console-consumer.sh --bootstrap-server 0.0.0.0:9092 --topic userMsg --group myConsumerGroup
msg1
查看myConsumerGroup的信息,此时消费者组中只有一个消费者,它一个消费所有分区。
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --list
myConsumerGroup
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
参数解释:
- TOPIC: 该消费者组消费的是哪些topic, 本例是只创建了一个topic, userMsg, 所以只消费了userMsg
- PARTITION: 表示该消费者消费的是哪些分区, 本例创建topic时有四个分区, 所以输出有四行, 分区号为0到3
- CURRENT-OFFSET: 表示消费者组最新消费的位移值, 此值在消费过程中是变化的
- LOG-END-OFFSET: 表示topic所有分区当前的日志终端位移值, 因为我们生产了1数据, 所以此处是1
- LAG: 表示滞后进度, 此值为LOG-END-OFFSET 与 CURRENT-OFFSET的差值, 代表的是滞后情况, 此值越大表示滞后严重, 本例LAG为0 说明没有消费滞后。
3. 此时消费者组中只有一个消费者。生产者继续生产消息。
生产者生产第二条消息msg2,查看myConsumerGroup消费者组:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
第三条消息:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
第四条消息:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 0 0 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 2 2 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 1 1 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
生产更多的消息:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 3 3 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 7 7 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
4. 再开启一个消费者,同样属于 myConsumerGroup。那么消费者组中有2个消费者。
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-console-consumer.sh --bootstrap-server 0.0.0.0:9092 --topic userMsg --group myConsumerGroup
查看myConsumerGroup,看到多了一个CONSUMER-ID. 说明这个消费者组中有2个消费者。
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 3 3 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 7 7 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
5. 生产2条消息,消费者组里的消费者各收到一条。
查看消费者组信息。消费2条之前:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 3 3 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 7 7 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
消费2条之后: 可以看到两个消费者各消费一条
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 4 4 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 8 8 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
6. 再开启2个消费者,依然属于myConsumerGroup.此时myConsumerGroup中有4个消费者。
查看此时的myConsumerGroup信息,能看到4个CONSUMER-ID,由于这个topic有4个partition,刚好每个consumer消费一个相应的partition:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 4 4 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-f4514659-6a5f-4ea0-b411-7cece8f37570 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 8 8 0 consumer-myConsumerGroup-1-cfcdac2d-9adb-47e1-8740-98cb1f8271cb /172.18.0.5 consumer-myConsumerGroup-1
这时生产4条消息,可以看到这4个消费者各自消费了一条,因为它们属于同一个消费者组。
7. 再开启1个消费者,同属于myConsumerGroup。那么一共有5个消费者属于这个消费者组。
查看消费者组信息,可以看到增加消费者之后,虽然CONSUMER-ID发生了变化,但是依然是4个消费者对应4个分区:
#增加第5个消费者之前:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 4 4 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 5 5 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 5 5 0 consumer-myConsumerGroup-1-f4514659-6a5f-4ea0-b411-7cece8f37570 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 8 8 0 consumer-myConsumerGroup-1-cfcdac2d-9adb-47e1-8740-98cb1f8271cb /172.18.0.5 consumer-myConsumerGroup-1
#增加第5个消费者之后:
bash-4.4# /opt/kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh -bootstrap-server 0.0.0.0:9092 --describe --group myConsumerGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
myConsumerGroup userMsg 0 5 5 0 consumer-myConsumerGroup-1-033a09ab-9942-473d-9d54-386b27718725 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 1 6 6 0 consumer-myConsumerGroup-1-48c4f84d-f40a-4d15-b0aa-071c39c3d4f5 /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 3 6 6 0 consumer-myConsumerGroup-1-cfcdac2d-9adb-47e1-8740-98cb1f8271cb /172.18.0.5 consumer-myConsumerGroup-1
myConsumerGroup userMsg 2 9 9 0 consumer-myConsumerGroup-1-bf6febff-32fa-496b-99c2-eed134235dc4 /172.18.0.5 consumer-myConsumerGroup-1
此时生产者生产十多条消息,可以看到有一个消费者一直消费不到消息。这是因为partition个数为4,而消费者组中的消费者个数为5,超过了分区数,造成浪费了一个消费者资源。
8. 消费者组总结
多个消费者可以通过拥有同样的group id,组成一个消费者组。
一个消费者组中的消费者共同消费消息,某一个topic的同一消息只能被其中一个消费者消费。