在Kafka中，当有新消费者加入或者订阅的topic数发生变化时，会触发Rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下：
第一步：所有成员都向coordinator发送请求，请求入组。一旦所有成员都发送了请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader。
第二步：leader开始分配消费方案，指明具体哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案发给coordinator。coordinator接收到分配方案之后会把方案发给各个consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。
所以对于Rebalance来说，Coordinator起着至关重要的作用，那么怎么查看消费者对应的Coordinator呢，我们知道某个消费者组对应__consumer_offsets中的哪个Partation是通过hash计算出来的：partation=hash("test_group_1")%50=28，表示test_group_1这个消费者组属于28号partation，通过命令:
./kafka-topics.sh --zookeeper 192.168.33.11: --describe --topic __consumer_offsets
可以找到28号Partation所对应的信息：
从而可以知道coordinator对应的broker为1
在Rebalance期间，消费者会出现无法读取消息，造成整个消费者群组一段时间内不可用，假设现在消费者组当中有A，代码逻辑执行10s，如果消费者组在消费的过程中consumer B加入到了该消费者组，并且B的代码逻辑执行20s，那么当A处理完后先进入Rebalance状态等待，只有当B也处理完后，A和B才真正通过Rebalance重新分配，这样显然A在等待的过程中浪费了资源。
消费者A：
"""
consumer_rebalance_a.py a消费者
"""
import pickle
import uuid
import time
from kafka import KafkaConsumer
from kafka.structs import TopicPartition, OffsetAndMetadata
from kafka import ConsumerRebalanceListener consumer = KafkaConsumer(
bootstrap_servers=['192.168.33.11:9092'],
group_id="test_group_1",
client_id="{}".format(str(uuid.uuid4())),
enable_auto_commit=False,
key_deserializer=lambda k: pickle.loads(k),
value_deserializer=lambda v: pickle.loads(v)
) # 用来记录最新的偏移量信息.
consumer_offsets = {} class MineConsumerRebalanceListener(ConsumerRebalanceListener):
def on_partitions_revoked(self, revoked):
"""
再均衡开始之前 下一轮poll之前触发
:param revoked:
:return:
"""
print('再均衡开始之前被自动触发.')
print(revoked, type(revoked))
consumer.commit_async(offsets=consumer_offsets) def on_partitions_assigned(self, assigned):
"""
再均衡完成之后 即将下一轮poll之前 触发
:param assigned:
:return:
"""
print('在均衡完成之后自动触发.')
print(assigned, type(assigned)) consumer.subscribe(topics=('round_topic',), listener=MineConsumerRebalanceListener()) def _on_send_response(*args, **kwargs):
"""
提交偏移量涉及回调函数
:param args:
:param kwargs:
:return:
"""
if isinstance(args[1], Exception):
print('偏移量提交异常. {}'.format(args[1]))
else:
print('偏移量提交成功') try:
start_time = time.time()
while True:
# 再均衡其实是在poll之前完成的
consumer_records_dict = consumer.poll(timeout_ms=100) # 处理逻辑.
for k, record_list in consumer_records_dict.items():
for record in record_list:
print("topic = {},partition = {},offset = {},key = {},value = {}".format(
record.topic, record.partition, record.offset, record.key, record.value)
) consumer_offsets[
TopicPartition(record.topic, record.partition)
] = OffsetAndMetadata(
record.offset + 1, metadata='偏移量.'
) try:
consumer.commit_async(callback=_on_send_response)
time.sleep(10)
except Exception as e:
print('commit failed', str(e)) except Exception as e:
print(str(e))
finally:
try:
# 同步提交偏移量,在消费者异常退出的时候再次提交偏移量,确保偏移量的提交.
consumer.commit()
print("同步补救提交成功")
except Exception as e:
consumer.close()
消费者B：
"""
consumer b.py 消费者B
""" import pickle
import uuid
import time
from kafka import KafkaConsumer
from kafka.structs import TopicPartition, OffsetAndMetadata
from kafka import ConsumerRebalanceListener consumer = KafkaConsumer(
bootstrap_servers=['192.168.33.11:9092'],
group_id="test_group_1",
client_id="{}".format(str(uuid.uuid4())),
enable_auto_commit=False, # 设置为手动提交偏移量.
key_deserializer=lambda k: pickle.loads(k),
value_deserializer=lambda v: pickle.loads(v)
) consumer_offsets = {} # 用来记录最新的偏移量信息. class MineConsumerRebalanceListener(ConsumerRebalanceListener):
def on_partitions_revoked(self, revoked):
"""
再均衡开始之前 下一轮poll之前触发
:param revoked:
:return:
"""
print('再均衡开始之前被自动触发.')
print(revoked, type(revoked))
consumer.commit_async(offsets=consumer_offsets) def on_partitions_assigned(self, assigned):
"""
再均衡完成之后 即将下一轮poll之前 触发
:param assigned:
:return:
""" print('在均衡完成之后自动触发.')
print(assigned, type(assigned)) consumer.subscribe(topics=('round_topic',), listener=MineConsumerRebalanceListener()) def _on_send_response(*args, **kwargs):
"""
提交偏移量涉及回调函数
:param args:
:param kwargs:
:return:
""" if isinstance(args[1], Exception):
print('偏移量提交异常. {}'.format(args[1]))
else:
print('偏移量提交成功') try:
start_time = time.time()
while True:
# 再均衡其实是在poll之前完成的
consumer_records_dict = consumer.poll(timeout_ms=100) record_num = 0
for key, record_list in consumer_records_dict.items():
for record in record_list:
record_num += 1
print("---->当前批次获取到的消息个数是:{}".format(record_num)) # 处理逻辑.
for k, record_list in consumer_records_dict.items():
for record in record_list:
print("topic = {},partition = {},offset = {},key = {},value = {}".format(
record.topic, record.partition, record.offset, record.key, record.value)
) consumer_offsets[
TopicPartition(record.topic, record.partition)
] = OffsetAndMetadata(record.offset + 1, metadata='偏移量.') try:
# 轮询一个batch 手动提交一次
consumer.commit_async(callback=_on_send_response)
time.sleep(20)
except Exception as e:
print('commit failed', str(e)) except Exception as e:
print(str(e))
finally:
try:
# 同步提交偏移量,在消费者异常退出的时候再次提交偏移量,确保偏移量的提交.
consumer.commit()
print("同步补救提交成功")
except Exception as e:
consumer.close()
消费者A和消费者B是同一个消费者组(test_group_1)的两个消费者，用time.sleep的方式模拟执行时间，A：10s，B：20s；首先A开始消费，当B新加入消费者组的时候会触发Rebalance，可以通过实现再均衡监听器(RebalanceListener)中的on_partitions_revoked和on_partitions_assigned方法来查看再均衡触发前后的partition变化情况，依次启动消费者A和B之后：
消费者A：
再均衡开始之前被自动触发.
{TopicPartition(topic='round_topic', partition=0), TopicPartition(topic='round_topic', partition=1), TopicPartition(topic='round_topic', partition=2)} <class 'set'>
<----------------------------------------
---------------------------------------->
在均衡完成之后自动触发.
{TopicPartition(topic='round_topic', partition=0), TopicPartition(topic='round_topic', partition=1)} <class 'set'>
<---------------------------------------- 消费者B：
再均衡开始之前被自动触发.
set() <class 'set'>
<----------------------------------------
---------------------------------------->
在均衡完成之后自动触发.
{TopicPartition(topic='round_topic', partition=2)} <class 'set'>
<----------------------------------------
在等待B的逻辑执行完后，A和B进入再均衡状态；再均衡前A处于partition 0、1、 2三个分区，B不占有任何partition；当再均衡结束后，A占有partition 0、1，B占有partition 2；然后A和B分别开始消费对应的partition。
在上述消费者A和B的代码中重写了RebalanceListener，主要是为了在发生再均衡之前提交最后一个已经处理记录的偏移量，因为再均衡时消费者将失去对一个分区的所有权，如果消费者已经消费了当前partition还没提交offset，这时候发生再均衡会使得消费者重新分配partition，可能使得同一个消息先后被两个消费者消费的情况，实现MineConsumerRebalanceListener再均衡前提交一次offset，确保每一个消费者在触发再均衡前提交最后一次offset：
class MineConsumerRebalanceListener(ConsumerRebalanceListener):
def on_partitions_revoked(self, revoked):
"""
再均衡开始之前 下一轮poll之前触发
:param revoked:
:return:
"""
print('再均衡开始之前被自动触发.')
print(revoked, type(revoked))
consumer.commit_async(offsets=consumer_offsets) def on_partitions_assigned(self, assigned):
"""
再均衡完成之后 即将下一轮poll之前 触发
:param assigned:
:return:
""" print('在均衡完成之后自动触发.')
print(assigned, type(assigned))
再均衡发生的场景有以下几种：
1. 组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了)
2. 订阅主题数发生变更，如果你使用了正则表达式的方式进行订阅，那么新建匹配正则表达式的topic就会触发rebalance
3. 订阅主题的分区数发生变更
鉴于触发再均衡后会造成资源浪费的问题，所以我们尽量不要触发再均衡