kafka消费者心跳线程

boolean ensureActiveGroup(final Timer timer) {
//...
//启动心跳
startHeartbeatThreadIfNeeded();
return joinGroupIfNeeded(timer);
}
//如果不存在心跳，创建。所以这里是加锁进行判断的，保证线程安全
private synchronized void startHeartbeatThreadIfNeeded() {
if (heartbeatThread == null) {
heartbeatThread = new HeartbeatThread();
heartbeatThread.start();
}
}

private class HeartbeatThread extends KafkaThread implements AutoCloseable {
//是否允许心跳标志位，写入要加锁
private boolean enabled = false;
//是否关闭标志位，写入要加锁
private boolean closed = false;
//心跳失败的原因，初始化为null，使用原子整型
private final AtomicReference failed = new AtomicReference<>(null);
}

public void run() {
//...
//主循环体，心跳线程是一个死循环
while (true) {
synchronized (AbstractCoordinator.this) {
//是否关闭
if (closed)
return;
//是否禁止
if (!enabled) {
AbstractCoordinator.this.wait();
continue;
}
//状态不对或者心跳失败
if (state.hasNotJoinedGroup() || hasFailed()) {
disable();
continue;
}
//唤醒客服端
client.pollNoWakeup();
long now = time.milliseconds();
//协调者未知
if (coordinatorUnknown()) {
if (findCoordinatorFuture != null) {
clearFindCoordinatorFuture();
AbstractCoordinator.this.wait(rebalanceConfig.retryBackoffMs);
} else {
lookupCoordinator();
}
//获取服务端心跳响应超时，置为协调者未知，等待下一轮
} else if (heartbeat.sessionTimeoutExpired(now)) {
markCoordinatorUnknown("session timed out without receiving a "
+ "heartbeat response");
//心跳超时，处理上次消息的时间过长，导致长时间没有poll
} else if (heartbeat.pollTimeoutExpired(now)) {
//发送离开请求，这个会导致reblance
maybeLeaveGroup(leaveReason);
//是否到了心跳时间
} else if (!heartbeat.shouldHeartbeat(now)) {
//阻塞到下一次重试
AbstractCoordinator.this.wait(rebalanceConfig.retryBackoffMs);
} else {
//设置心跳的时间为当前时间，这个是更新本地心跳的时间
heartbeat.sentHeartbeat(now);
//发送心跳请求并获取异步future对象
final RequestFuture<Void> heartbeatFuture = sendHeartbeatRequest();
//添加监听器
heartbeatFuture.addListener(new RequestFutureListener<Void>() {
@Override
//心跳成功
public void onSuccess(Void value) {
synchronized (AbstractCoordinator.this) {
heartbeat.receiveHeartbeat();
}
}

@Override
public void onFailure(RuntimeException e) {
synchronized (AbstractCoordinator.this) {
//正在reblance当做正常响应
if (e instanceof RebalanceInProgressException) {
heartbeat.receiveHeartbeat();
} else if (e instanceof FencedInstanceIdException) {
heartbeatThread.failed.set(e);
} else {
heartbeat.failHeartbeat();
// 唤醒线程，等待下一轮
AbstractCoordinator.this.notify();
}
}
}
});
}
}
}
}
//...
//异常处理逻辑
}

synchronized RequestFuture sendHeartbeatRequest() {
//组装心跳消息，包含消费者组信息，消费者信息
HeartbeatRequest.Builder requestBuilder =
new HeartbeatRequest.Builder(new HeartbeatRequestData()
.setGroupId(rebalanceConfig.groupId)
.setMemberId(this.generation.memberId)
.setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
.setGenerationId(this.generation.generationId));
//发送消息
return client.send(coordinator, requestBuilder)
.compose(new HeartbeatResponseHandler(generation));
}
处理结果是用一个新的类HeartbeatResponseHandler来实现，现在可以看下这个类的handle方法：
public void handle(HeartbeatResponse heartbeatResponse, RequestFuture future) {
//处理的异常
Errors error = heartbeatResponse.error();
//没有异常，future成功返回
if (error == Errors.NONE) {
future.complete(null);
//协调者未知
} else if (error == Errors.COORDINATOR_NOT_AVAILABLE
|| error == Errors.NOT_COORDINATOR) {
markCoordinatorUnknown(error);
future.raise(error);
//正在reblance
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
if (state == MemberState.STABLE) {
//如果是稳定状态，请求重新加入
requestRejoin();
future.raise(error);
} else {
//如果不是stable状态，忽视
future.complete(null);
}
//leader版本已经替换
} else if (error == Errors.ILLEGAL_GENERATION ||
error == Errors.UNKNOWN_MEMBER_ID ||
error == Errors.FENCED_INSTANCE_ID) {
if (generationUnchanged()) {
resetGenerationOnResponseError(ApiKeys.HEARTBEAT, error);
future.raise(error);
} else {
future.complete(null);
}
//没权限
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
future.raise(GroupAuthorizationException.forGroupId(rebalanceConfig.groupId));
} else {
//其他异常
future.raise(new KafkaException("Unexpected error in heartbeat response: " + error.message()));
}
}