KafkaConsumer HeartbeatThread

平凡的学生族

发布于 2021-12-08 14:20:22

3250

发布于 2021-12-08 14:20:22

文章被收录于专栏：后端技术

背景

我们已知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);
}