Kafka27:消费者Range分配

Range 以及再平衡

1. Range 分区策略原理

• 注意：
• 如果只是针对1 个topic 而言，C0消费者多消费1个分区影响不是很大。但是如果有N 多个topic，那么针对每个topic，消费者C0都将多消费1 个分区，topic越多，C0消费的分区会比其他消费者明显多消费N 个分区。
• 容易产生数据倾斜！

2. Range 分区策略案例

1. 修改主题first 为7 个分区

# 增加first主题的分区数
root@redis01:/usr/local/kafka# bin/kafka-topics.sh --bootstrap-server redis01:9092 --alter --topic first --partitions 7

# 查看
root@redis01:/usr/local/kafka# bin/kafka-topics.sh --bootstrap-server redis01:9092 --describe --topic first

1. 创建消费者

• 注意：partition_assignment_strategy=[RangePartitionAssignor] ，需要导入from kafka.coordinator.assignors.range import RangePartitionAssignor

• 小插曲：python-Kafka模块测试失败，我在这里这样partition_assignment_strategy=['RangePartitionAssignor'] 导入，一直报错，让我花了好久一会儿，我原本认为是python-Kafka库已经设置好了字符转化，原来需要自己from kafka.coordinator.assignors.range import RangePartitionAssignor使用。🤣令我高兴的是，尝试了另一个Python的Kafka库confluent_kafka 。😁

'''
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_group2",
        partition_assignment_strategy=[RangePartitionAssignor], )
    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()

1. 创建生成者

• 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()

1. 查看

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

3. Range分区分配再平衡案例

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

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

另一个库：confluent_kafka

librdkafka/CONFIGURATION.md at master · edenhill/librdkafka

socket.timeout.ms * 10 .. 300000 60000 low Default timeout for network requests. Producer: ProduceRequests will use the lesser value of socket.timeout.ms and remaining message.timeout.ms for the first message in the batch. Consumer: FetchRequests will use fetch.wait.max.ms + socket.timeout.ms. Admin: Admin requests will use socket.timeout.ms or explicitly set rd_kafka_AdminOptions_set_operation_timeout() value.

https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md