Kafka29:消费者Sticky分配

Sticky分区策略原理

粘性分区定义：

• 可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。

• 粘性分区是Kafka 从 0.11.x 版本开始引入这种分配策略 首先会尽量均衡的放置分区到消费者上面 ，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

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_group6",
        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()

注意：关于Kafka-Python库报错

• 使用

解决

• 找到你pip安装Kafka-Python库的 kafka/coordinator/assignors/sticky/sticky_assignor.py

• 将six.iteritems 修改为six.viewitems ，保存即可。

创建生成者

• 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后）

另一个库：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