GroupCoordinator介绍

GroupCoodinator 和KafkaController一样，在每一个broker都会启动一个GroupCoodinator，Kafka 按照消费者组的名称将其分配给对应的GroupCoodinator进行管理；每一个GroupCoodinator只负责管理一部分消费者组，而非集群中全部的消费者组。

一功能

# 记录消费者组的信息

# 通过心跳消息检测消费者状态

# 针对JoinGroupRequest和SyncGroupRequest进行消费者组中分区的分配

# 维护消费的offset信息

二 MemberMetadata

GroupCoodinator使用MemberMetadata记录消费者的元数据，包括以下字段：
memberId: String 对应消费者的id

groupId: String 消费者对应的消费者组的id

rebalanceTimeoutMs: Int rebalance操作超时时间

sessionTimeoutMs: Int 心跳超时时间

assignment:Array[Byte] 记录分配给当前消费者的分区消息

supportedProtocols: List[(String Array[Byte])] 消费者支持的PartitionAssignor

latestHeartbeat: Long 最后一次心跳消息的时间戳

isLeaving: Boolean 对应消费者是否已经离开消费者组了

三 GroupMetadata

GroupMetadata记录了消费者组的元数据信息，其字段含义如下：

groupId: String 消费者组对应的id

state: GroupState 消费者组当前所处的状态

members: HashMap[String, MemberMetadata] 保存了消费者和对应的消费者元数据信息

offsets: HashMap[TopicPartition,OffsetAndMetadata] 保存了分区和其对应的offset映射关系

generationId: Int 消费者组的年代信息

leaderId: String 消费者组中的leader信息

protocol: String 记录了当前消费者组选择的PartitionAssignor

它提供了一些方法，比如针对members的增删，对state的切换。在进行Member增删的时候，顺便会选择消费者组里leader

def add(memberId: String, member: MemberMetadata) {
if (members.isEmpty)
this.protocolType = Some(member.protocolType)

assert(groupId == member.groupId)
assert(this.protocolType.orNull == member.protocolType)
assert(supportsProtocols(member.protocols))

if (leaderId == null)
leaderId = memberId
members.put(memberId, member)
}

def remove(memberId: String) {members.remove(memberId)if (memberId == leaderId) {leaderId = if (members.isEmpty) {null} else {members.keys.head}}
}

为消费者组选择合适的PartitionAssignor功能：

def selectProtocol

: String = {if (members.isEmpty)throw new IllegalStateException("Cannot select protocol for empty group")// 所有消费者都支持的协议作为候选协议val candidates = candidateProtocols// 每一个消费者都会通过vote方法进行投票，会为其supportedProtocols集合里的第一个PartitionAssignor投一票，最终将选择的票最多的PartitionAssignorval votes: List[(String, Int)] = allMemberMetadata.map(_.vote(candidates)).groupBy(identity).mapValues(_.size).toListvotes.maxBy(_._2)._1
}

GroupCoordinator介绍相关推荐

简单介绍互联网领域选择与营销方法
在我看来,互联网领域的选择是"安家",而营销方法的不同则表现了"定家"的方式多种多样,只有选对了,"家"才得以"安定". ...
常用开源协议介绍以及开源软件规范列表
1. 开源协议介绍 GPL: General Public License,开源项目最常用的许可证,衍生代码的分发需开源并且也要遵守此协议.该协议也有很多变种,不同变种要求会略微不同. MPL: MP ...
python：Json模块dumps、loads、dump、load介绍
20210831 https://www.cnblogs.com/bigtreei/p/10466518.html json dump dumps 区别 python:Json模块dumps.load ...
pytorch学习笔记（九）：PyTorch结构介绍
PyTorch结构介绍对PyTorch架构的粗浅理解,不能保证完全正确,但是希望可以从更高层次上对PyTorch上有个整体把握.水平有限,如有错误,欢迎指错,谢谢! 几个重要的类型和数值相关的 T ...
Python字节码介绍
了解 Python 字节码是什么,Python 如何使用它来执行你的代码,以及知道它是如何帮到你的. 如果你曾经编写过 Python,或者只是使用过 Python,你或许经常会看到 Python 源代 ...
Pytest - 使用介绍
1. 概述 pytest是一个非常成熟的全功能的Python测试框架,主要特点有以下几点: 1.简单灵活,容易上手,文档丰富: 2.支持参数化,可以细粒度地控制要测试的测试用例: 3.能够支持简单的单 ...
遗传算法的简单介绍以及模式定理的简单证明
遗传算法遗传算法(Genetic Algorithm,GA),最早是由美国的John holland在20世纪70年代提出.算法通过模拟达尔文生物进化论的自然选择以及遗传学机理的生物进化过程来搜 ...
k8s核心组件详细介绍教程（配超详细实例演示）
本文实验环境基于上篇文章手把手从零开始搭建k8s集群超详细教程本文根据B站课程云原生Java架构师的第一课K8s+Docker+KubeSphere+DevOps学习总结而来 k8s核心组件介绍 1 ...
Tensorrt一些优化技术介绍
Tensorrt一些优化技术介绍 Figure 1. A quantizable AveragePool layer (in blue) is fused with a DQ layer and a ...

GroupCoordinator介绍

GroupCoordinator介绍相关推荐

最新文章

热门文章