Kafka源码深度解析-序列7 -Consumer -coordinator协议与heartbeat实现原理
发布日期:2021-07-12 08:49:13 浏览次数:2 分类:技术文章

本文共 12311 字,大约阅读时间需要 41 分钟。

转自:

单线程的consumer

在前面我们讲过,KafkaProducer是线程安全的,同时其内部还有一个Sender,开了一个后台线程,不断从队列中取消息进行发送。 

而consumer,是一个纯粹的单线程程序,后面所讲的所有机制,包括coordinator,rebalance, heartbeat等,都是在这个单线程的poll函数里面完成的。也因此,在consumer的代码内部,没有锁的出现。

//客户端线程while (true) {         ConsumerRecords
records = consumer.poll(100); 。。。 }
  • 1
  • 2
  • 3
  • 4
  • 5

何为coordinator?

去Zookeeper依赖

在0.9以前的client api中,consumer是要依赖Zookeeper的。因为同一个consumer group中的所有consumer需要进行协同,进行下面所讲的rebalance。

但是因为zookeeper的“herd”与“split brain”,导致一个group里面,不同的consumer拥有了同一个partition,进而会引起消息的消费错乱。为此,在0.9中,不再用zookeeper,而是Kafka集群本身来进行consumer之间的同步。下面引自kafka设计的原文:

The current version of the high level consumer suffers from herd and split brain problems, where multiple consumers in a group run a distributed algorithm to agree on the same partition ownership decision. Due to different view of the zookeeper data, they run into conflicts that makes the rebalancing attempt fail. But there is no way for a consumer to verify if a rebalancing operation completed successfully on the entire group. This also leads to some potential bugs in the rebalancing logic, for example, https://issues.apache.org/jira/browse/KAFKA-242
  • 1

为什么在一个group内部,1个parition只能被1个consumer拥有?

我们知道,对于属于不同consumer group的consumers,可以消费同1个partition,从而实现Pub/Sub模式。

但是在一个group内部,是不允许多个consumer消费同一个partition的,这也就意味着,对于1个topic,1个group来说, 其partition数目 >= consumer个数。

比如对于1个topic,有4个partition,那么在一个group内部,最多只能有4个consumer。你加入更多的consumer,它们也不会分配到partition。

那为什么要做这个限制呢?原因在下面这篇文章中,有详细阐述: 

简单来说,一个是因为这样做就没办法保证同1个partition消息的时序;另1方面,Kafka的服务器,是每个topic的每个partition的每个consumer group对应1个offset,即(topic, partition, consumer_group_id) – offset。如果多个consumer并行消费同1个partition,那offset的confirm就会出问题。

知道了这个前提,下面我们就来分析partition的分配问题。

coordinator协议 / partition分配问题

问题的提出: 

给定一个topic,有4个partition: p0, p1, p2, p3, 一个group有3个consumer: c0, c1, c2。那么,如果按范围分配策略,分配结果是: 
c0: p0, c1: p1, c2: p2, p3 
如果按轮询分配策略: 
c0: p1, p3, c1: p1, c2: p2

那这整个分配过程是如何进行的呢?见下图所示: 

这里写图片描述

3步分配过程

步骤1:对于每1个consumer group,Kafka集群为其从broker集群中选择一个broker作为其coordinator。因此,第1步就是找到这个coordinator。

private Map
>> pollOnce(long timeout) { coordinator.ensureCoordinatorKnown(); //poll函数的第1行,就是寻找coordinator。如果没找到,就会一直阻塞在这里 ... } public void ensureCoordinatorKnown() { while (coordinatorUnknown()) { RequestFuture
future = sendGroupMetadataRequest(); client.poll(future); if (future.failed()) { if (future.isRetriable()) client.awaitMetadataUpdate(); else throw future.exception(); } } } private RequestFuture
sendGroupMetadataRequest() { Node node = this.client.leastLoadedNode(); if (node == null) { return RequestFuture.noBrokersAvailable(); } else { GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId); //向集群中负载最小的node,发送请求,询问这个group id对应的coordinator是谁 return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest) .compose(new RequestFutureAdapter
() { @Override public void onSuccess(ClientResponse response, RequestFuture
future) { handleGroupMetadataResponse(response, future); } }); } }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34

步骤2:找到coordinator之后,发送JoinGroup请求

private Map
>> pollOnce(long timeout) { coordinator.ensureCoordinatorKnown(); //步骤1:寻找coordinator if (subscriptions.partitionsAutoAssigned()) coordinator.ensurePartitionAssignment(); //步骤2+3: JoinGroup + SyncGroup public void ensureActiveGroup() { if (!needRejoin()) return; if (needsJoinPrepare) { onJoinPrepare(generation, memberId); needsJoinPrepare = false; } while (needRejoin()) { ensureCoordinatorKnown(); if (client.pendingRequestCount(this.coordinator) > 0) { client.awaitPendingRequests(this.coordinator); continue; } RequestFuture
future = performGroupJoin(); client.poll(future); if (future.succeeded()) { onJoinComplete(generation, memberId, protocol, future.value()); needsJoinPrepare = true; heartbeatTask.reset(); } else { RuntimeException exception = future.exception(); if (exception instanceof UnknownMemberIdException || exception instanceof RebalanceInProgressException || exception instanceof IllegalGenerationException) continue; else if (!future.isRetriable()) throw exception; time.sleep(retryBackoffMs); } } }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42

步骤3:JoinGroup返回之后,发送SyncGroup,得到自己所分配到的partition

private RequestFuture
performGroupJoin() { if (coordinatorUnknown()) return RequestFuture.coordinatorNotAvailable(); // send a join group request to the coordinator log.debug("(Re-)joining group {}", groupId); JoinGroupRequest request = new JoinGroupRequest( groupId, this.sessionTimeoutMs, this.memberId, protocolType(), metadata()); // create the request for the coordinator log.debug("Issuing request ({}: {}) to coordinator {}", ApiKeys.JOIN_GROUP, request, this.coordinator.id()); return client.send(coordinator, ApiKeys.JOIN_GROUP, request) .compose(new JoinGroupResponseHandler()); } private class JoinGroupResponseHandler extends CoordinatorResponseHandler
{ @Override public JoinGroupResponse parse(ClientResponse response) { return new JoinGroupResponse(response.responseBody()); } @Override public void handle(JoinGroupResponse joinResponse, RequestFuture
future) { // process the response short errorCode = joinResponse.errorCode(); if (errorCode == Errors.NONE.code()) { log.debug("Joined group: {}", joinResponse.toStruct()); AbstractCoordinator.this.memberId = joinResponse.memberId(); AbstractCoordinator.this.generation = joinResponse.generationId(); AbstractCoordinator.this.rejoinNeeded = false; AbstractCoordinator.this.protocol = joinResponse.groupProtocol(); sensors.joinLatency.record(response.requestLatencyMs()); if (joinResponse.isLeader()) { onJoinLeader(joinResponse).chain(future); //关键点:在JoinGroup返回之后,竟跟着发送SyncGroup消息 } else { onJoinFollower().chain(future); } } else if (errorCode == Errors.GROUP_LOAD_IN_PROGRESS.code()) { 。。。 } } }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48

注意,在上面3步中,有一个关键点: partition的分配策略和分配结果其实是由client决定的,而不是由coordinator决定的。什么意思呢?在第2步,所有consumer都往coordinator发送JoinGroup消息之后,coordinator会指定其中一个consumer作为leader,其他consumer作为follower。

然后由这个leader进行partition分配。然后在第3步,leader通过SyncGroup消息,把分配结果发给coordinator,其他consumer也发送SyncGroup消息,获得这个分配结果。

为什么要在consumer中选一个leader出来,进行分配,而不是由coordinator直接分配呢?关于这个, Kafka的官方文档有详细的分析。其中一个重要原因是为了灵活性:如果让server分配,一旦需要新的分配策略,server集群要重新部署,这对于已经在线上运行的集群来说,代价是很大的;而让client分配,server集群就不需要重新部署了。

rebalance机制

所谓rebalance,就是在某些条件下,partition要在consumer中重新分配。那哪些条件下,会触发rebalance呢? 

条件1:有新的consumer加入 
条件2:旧的consumer挂了 
条件3:coordinator挂了,集群选举出新的coordinator 
条件4:topic的partition新加 
条件5:consumer调用unsubscrible(),取消topic的订阅

当consumers检测到要rebalance时,所有consumer都会重走上面的流程,进行步骤2 + 步骤3: JoinGroup + SyncGroup。

可问题是: 当一个consumer挂了,或者有新的consumer加入,其他consumers怎么知道要进行rebalance呢? 答案就是下面的heartbeat。

heartbeat的实现原理

每一个consumer都会定期的往coordinator发送heartbeat消息,一旦coordinator返回了某个特定的error code:ILLEGAL_GENERATION, 就说明之前的group无效了(解散了),要重新进行JoinGroup + SyncGroup操作。

那这个定期发送如何实现呢?一个直观的想法就是开一个后台线程,定时发送heartbeat消息,但维护一个后台线程,很显然会增大实现的复杂性。上面也说了, consumer是单线程程序。在这里是通过DelayedQueue来实现的。

DelayedQueue与HeartBeatTask

其基本思路是把HeartBeatRequest放入一个DelayedQueue中,然后在while循环的poll中,每次从DelayedQueue中把请求拿出来发送出去(只有时间到了,Task才能从Queue中拿出来)。

private class HeartbeatTask implements DelayedTask { private boolean requestInFlight = false; //关键变量:判断当前是否有发送出去的HeartBeatRequest,其Response还没回来 //reset本质就是发送函数 public void reset() { long now = time.milliseconds(); heartbeat.resetSessionTimeout(now); client.unschedule(this); if (!requestInFlight) client.schedule(this, now); //如果没有requestInFlight,则把HeartBeatTask放入DelayedQueue中 } @Override public void run(final long now) { if (generation < 0 || needRejoin() || coordinatorUnknown()) { return; } if (heartbeat.sessionTimeoutExpired(now)) { coordinatorDead(); return; } if (!heartbeat.shouldHeartbeat(now)) { client.schedule(this, now + heartbeat.timeToNextHeartbeat(now)); } else { heartbeat.sentHeartbeat(now); requestInFlight = true; RequestFuture
future = sendHeartbeatRequest(); future.addListener(new RequestFutureListener
() { @Override public void onSuccess(Void value) { requestInFlight = false; long now = time.milliseconds(); heartbeat.receiveHeartbeat(now); long nextHeartbeatTime = now + heartbeat.timeToNextHeartbeat(now); //放入delayedQueue client.schedule(HeartbeatTask.this, nextHeartbeatTime); } //hearbeat返回之后,无论response成功,还是失败,把下1个heartbeat放入delayedQueue,从而形成循环间隔发送 @Override public void onFailure(RuntimeException e) { requestInFlight = false; client.schedule(HeartbeatTask.this, time.milliseconds() + retryBackoffMs); } }); } }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

rebalance检测

首先一点说明:个人认为这里的网络框架,封装的有点冗余:sendHeartbeatRequest既有callback机制(CompleteHandler),又为其Future加了Listener机制(上面的代码)。

也就是在heartbeat的completeHandler中,完成了rebalance的检测:从下面代码可以看出,对于以下的response error code,都会触发rebalance:

* GROUP_COORDINATOR_NOT_AVAILABLE (15) * NOT_COORDINATOR_FOR_GROUP (16) * ILLEGAL_GENERATION (22) * UNKNOWN_MEMBER_ID (25) * REBALANCE_IN_PROGRESS (27) * GROUP_AUTHORIZATION_FAILED (30)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
public RequestFuture
sendHeartbeatRequest() { HeartbeatRequest req = new HeartbeatRequest(this.groupId, this.generation, this.memberId); return client.send(coordinator, ApiKeys.HEARTBEAT, req) .compose(new HeartbeatCompletionHandler()); } private class HeartbeatCompletionHandler extends CoordinatorResponseHandler
{ @Override public HeartbeatResponse parse(ClientResponse response) { return new HeartbeatResponse(response.responseBody()); } @Override public void handle(HeartbeatResponse heartbeatResponse, RequestFuture
future) { sensors.heartbeatLatency.record(response.requestLatencyMs()); short errorCode = heartbeatResponse.errorCode(); if (errorCode == Errors.NONE.code()) { log.debug("Received successful heartbeat response."); future.complete(null); } else if (errorCode == Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code() || errorCode == Errors.NOT_COORDINATOR_FOR_GROUP.code()) { log.info("Attempt to heart beat failed since coordinator is either not started or not valid, marking it as dead."); coordinatorDead(); future.raise(Errors.forCode(errorCode)); } else if (errorCode == Errors.REBALANCE_IN_PROGRESS.code()) { log.info("Attempt to heart beat failed since the group is rebalancing, try to re-join group."); AbstractCoordinator.this.rejoinNeeded = true; future.raise(Errors.REBALANCE_IN_PROGRESS); } else if (errorCode == Errors.ILLEGAL_GENERATION.code()) { log.info("Attempt to heart beat failed since generation id is not legal, try to re-join group."); AbstractCoordinator.this.rejoinNeeded = true; future.raise(Errors.ILLEGAL_GENERATION); } else if (errorCode == Errors.UNKNOWN_MEMBER_ID.code()) { log.info("Attempt to heart beat failed since member id is not valid, reset it and try to re-join group."); memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID; AbstractCoordinator.this.rejoinNeeded = true; future.raise(Errors.UNKNOWN_MEMBER_ID); } else if (errorCode == Errors.GROUP_AUTHORIZATION_FAILED.code()) { future.raise(new GroupAuthorizationException(groupId)); } else { future.raise(new KafkaException("Unexpected errorCode in heartbeat response: " + Errors.forCode(errorCode).exception().getMessage())); } } }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45

关键点:这里所谓的触发,其实就是把rejoinNeeded置为了true。然后在下一次poll循环的时候,检测到rejoinNeeded为true,就会重走上面的步骤2 + 步骤3

failover

对于这整个体系来说,consumer可能挂掉,coordinator也可能挂掉。因此双方需要互相检测,对方是否挂了。

检测方法同样是上面的heartbeat:当consumer发现heartbeat返回超时,或者coordinator很久没有收到heartbeat,就会认为对方挂了。

当然,这会有“误操作”,比如consumer处理消息很慢(因为是单线程),导致下1次heartbeat迟迟没有发出去。这个时候coordinator就会认为该consumer挂了,会主动断开连接。从而触发1次rebalance。

consumer认为coordinator挂了

就会从上面的步骤1开始,重新discovery coordinator,然后JoinGroup + SyncGroup

coordinator认为consumer挂了

从上面的步骤2开始,通知其他所有剩下的consumers,进行JoinGroup + SyncGroup

转载于:https://www.cnblogs.com/wq3435/p/7977843.html

转载地址:https://blog.csdn.net/ankan7400/article/details/102088926 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!

上一篇:apache kafka系列之在zookeeper中存储结构
下一篇:Kafka设计解析(一)- Kafka背景及架构介绍

发表评论

最新留言

路过按个爪印,很不错,赞一个!
[***.219.124.196]2024年03月17日 08时03分13秒

关于作者

    喝酒易醉,品茶养心,人生如梦,品茶悟道,何以解忧?唯有杜康!
-- 愿君每日到此一游!

推荐文章

linux修改文件权限为所有人都可以访问,Linux 笔记分享八:文件权限的设定 2019-04-21
linux中卸载ambri-servle,Kerberos 命令使用 2019-04-21
linux二进制反编译,Xori:一款来自BlackHat 2018的二进制反汇编和静态分析工具 2019-04-21
linux两台主机添加信任,Linux两台机器间添加信任,实现不用密码问,互传文件... 2019-04-21
linux 自动获取ssl证书,linux生成自验证ssl证书的具体命令和步骤 2019-04-21
linux基础命令20个,20-linux中基础命令 2019-04-21
重置网络配置 android,重置Android移动网络信号? 2019-04-21
android 6.0权限统一,snails-permission 2019-04-21
html压缩包用什么软件打开方式,分享一招超简单打开zip文件的方法 2019-04-21
html5 拍照 分辨率,getUserMedia API及HTML5 调用手机摄像头拍照 2019-04-21
dubbo全局异常处理_基于spring aop的dubbo异常统一处理 2019-04-21
java 程序找错_java代码找错 2019-04-21
Java线程怎么发送消息_Java客户端Socket如何能在阻塞线程下收到服务端发送来的消息?... 2019-04-21
网页无法调用java9_JAVA 9 (内部类,异常,包) 2019-04-21
java约瑟夫环pta上_cdoj525-猴子选大王 (约瑟夫环) 2019-04-21
java++记录+运行_记录java+testng运行selenium(三)---xml、ini、excel、日志等配置 2019-04-21
mysql居左查询abcd_MySql速查手册 2019-04-21
loadrunner 错误: 无法找到 java.exe_LoadRunner错误及解决方法总结 2019-04-21
Java小魔女芭芭拉_沉迷蘑菇不可自拔,黏土人《小魔女学园》苏西·曼芭芭拉 图赏... 2019-04-21
php+mysql记事本_一个简单记事本php操作mysql辅助类创建 2019-04-21