背景
我们已知ConsumerNetworkClient是对NetworkClient的封装，而消费者与服务器的交互分为两条线程
- KafkaConsumer.poll() 业务线程
- HeartbeatThread 心跳线程
初始化
在激活消费者组状态时，会启动心跳线程。
心跳线程启动后，只要没销毁，就会继续执行，期间可能被沉默，也可能由于父对象的coordinator变化而改变发送心跳对象。
AbstractCoordinator:
public void ensureActiveGroup() {
// always ensure that the coordinator is ready because we may have been disconnected
// when sending heartbeats and does not necessarily require us to rejoin the group.
ensureCoordinatorReady();
startHeartbeatThreadIfNeeded();
joinGroupIfNeeded();
}
private synchronized void startHeartbeatThreadIfNeeded() {
if (heartbeatThread == null) {
heartbeatThread = new HeartbeatThread();
heartbeatThread.start();
}
}
发送心跳
向父对象的coordinator发送心跳。
// visible for testing
synchronized RequestFuture<Void> sendHeartbeatRequest() {
log.debug("Sending Heartbeat request to coordinator {}", coordinator);
HeartbeatRequest.Builder requestBuilder =
new HeartbeatRequest.Builder(this.groupId, this.generation.generationId, this.generation.memberId);
return client.send(coordinator, requestBuilder)
.compose(new HeartbeatResponseHandler());
}
心跳间歇
HeartbeatThread主循环在判断下次心跳时间未到时会阻塞。
else if (!heartbeat.shouldHeartbeat(now)) {
// poll again after waiting for the retry backoff in case the heartbeat failed or the
// coordinator disconnected
AbstractCoordinator.this.wait(retryBackoffMs);
}
等待时间为heartbeatInterval，但为什么等待retryBackoffMs？
