Kafka28:消费者Roundrobin分配

RoundRobin 分区策略原理

• RoundRobin 针对集群中所有Topic而言。

 

• RoundRobin 轮询分区策略，是把所有的partition 和所有的consumer 都列出来，然后按照hashcode 进行排序，最后通过轮询算法来分配partition 给到各个消费者。

RoundRobin 分区分配策略案例

创建消费者

• 依次在Consumer1、Consumer2、Consumer3三个消费者代码中修改分区分配策略为RoundRobin。

'''
Description: henggao_note
version: v1.0.0
Author: henggao
Date: 2022-03-03 23:18:41
LastEditors: henggao
LastEditTime: 2022-03-05 10:54:35
'''

from kafka import KafkaConsumer
from kafka.coordinator.assignors.range import RangePartitionAssignor
from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
# from kafka.coordinator.assignors.sticky.sticky_assignor import  StickyPartitionAssignor
import time

# topic_name = 'my_favorite_topic2'
topic_name = 'first'


def start_consumer():
    consumer = KafkaConsumer(topic_name, bootstrap_servers=[
        '192.168.92.145:9092', '192.168.92.146:9092', '192.168.92.147:9092', '192.168.92.148:9092'], group_id="my_group3",
        partition_assignment_strategy=[RoundRobinPartitionAssignor], )
    for msg in consumer:
        # print(msg)
        # print("topic = %s" % msg.topic)  # topic default is string
        # print("partition = %d" % msg.partition)
        # print("value = %s" % msg.value.decode())  # bytes to string
        # print("time = ", time.strftime("%Y-%m-%d %H:%M:%S",
        #       time.localtime(msg.timestamp/1000)))
        print("topic = %s" % msg.topic + ", partition = %d" %
              msg.partition + ", value = %s" % msg.value.decode())  # topic default is string


if __name__ == '__main__':
    start_consumer()

创建生成者

• producter.py ，发送500条数据，随机发送到不同的分区。

'''
Description: henggao_learning
version: v1.0.0
Author: henggao
Date: 2022-02-28 10:11:09
LastEditors: henggao
LastEditTime: 2022-03-04 09:31:49
'''
from concurrent.futures import thread
from kafka import KafkaProducer
from time import sleep, thread_time

# topic_name = 'kafka_demo'
topic_name = 'first'


def start_producer():
    producer = KafkaProducer(bootstrap_servers=[
        '192.168.92.145:9092', '192.168.92.146:9092', '192.168.92.147:9092', '192.168.92.148:9092'])
    for i in range(0, 500):
        msg = 'msg_test is ' + str(i)
        # producer.send(topic_name, msg.encode('utf-8') ,partition=0)
        producer.send(topic_name, msg.encode('utf-8'))
        sleep(0.03)


if __name__ == '__main__':
    start_producer()

查看

• 可以发现consumer1.py、consumer2.py、consumer3.py ，3 个消费者分别消费哪些分区的数据

场景一：Consumer1挂掉一段时间内（45s以内）

场景二：Consumer1挂掉一段时间后（45s后）