Kafka2.0消费者客户端源码分析

1 KafkaConsumer 构造器

1. 初始化参数配置。
2. 初始化消费者网络客户端ConsumerNetworkClient。
3. 初始化消费者协调器ConsumerCoordinator

4. 初始化拉取器Fetcher

   2 订阅主题

5. 调用订阅方法subscribe()、assign()会将订阅信息记录到SubscriptionState，屡次订阅会覆盖旧数据。

6. 若是元数据缓存Metadata不包含订阅的主题，则设置needUpdate=true，标识须要更新元数据。

   3 拉取消息

7. poll()方法指定超时时间timeoutMs，在这个时间范围内不断轮询。
8. 更新分配给消费者的数据，包括消费者协调器、偏移量、心跳等。
9. 根据超时时间拉取消息。
10. 若是拉取的消息不为空，当即出发下一轮的拉取，能够避免因处理消息响应，而阻塞等待。
11. 拉取的消息会先反序列化，再调用消费者拦截器，最后返回给消费者。

12. 拉取超时后，返回空记录。

private ConsumerRecords<K, V> poll(final long timeoutMs, final boolean includeMetadataInTimeout) {
    long elapsedTime = 0L;
    do {
        final long metadataEnd;
        // 更新分配元数据，协调器、心跳、消费位置
        if (!updateAssignmentMetadataIfNeeded(remainingTimeAtLeastZero(timeoutMs, elapsedTime))) {
            return ConsumerRecords.empty();
        }
        // 拉取消息
        final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(remainingTimeAtLeastZero(timeoutMs, elapsedTime));

        if (!records.isEmpty()) {
            // 消息不为空时，当即发起下一轮的拉取消息，避免阻塞等待响应处理。
            // 注意，在消息返回以前，不能触发唤醒或其余错误。
            if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {
                client.pollNoWakeup();
            }
            // 回调执行消费者拦截器后返回给消费者
            return this.interceptors.onConsume(new ConsumerRecords<>(records));
        }
        final long fetchEnd = time.milliseconds();
        elapsedTime += fetchEnd - metadataEnd;
    } while (elapsedTime < timeoutMs); // 轮询拉取，知道超过输入的超时时间

    return ConsumerRecords.empty();
}

3.1 拉取消息的详细流程

1. 若是分区记录缓存PartitionRecords存在缓存记录，则优先会从分区记录缓存队列completedFetches中拉取一部分记录，直接返回。
2. 不然，向服务端发送拉取请求，消费者并不会当即发送请求，而是先构造 Node 和请求的缓存LinkedHashMap。
3. 遍历上述缓存，构形成能够直接发送的请求，并缓存到ConcurrentHashMap<Node, ConcurrentLinkedQueue<ClientRequest>> unsent中，同时添加处理响应的监听器。
4. 遍历unsent，使用NetworkClient发送请求，这里和生产者使用相同的方法，处理流程类似。发送完后即清空unsent。
5. 当拉取到消息，会回调第3步中的监听器，将消息缓存到队列ConcurrentLinkedQueue<CompletedFetch> completedFetches。
6. 相似第1步，从分区记录缓存队列completedFetches中拉取一部分记录返回给消费者。

private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(final long timeoutMs) {
    final long startMs = time.milliseconds();
    long pollTimeout = Math.min(coordinator.timeToNextPoll(startMs), timeoutMs);

    final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords(); // 从缓存队列拉取
    if (!records.isEmpty()) { // 缓存中有数据则直接返回
        return records;
    }

    // 1.将拉取请求构形成节点和请求的映射关系，并缓存在 unsent
    // 2.添加响应处理监听器，处理发送拉取请求后，从服务端返回的消息，并缓存在队列中
    fetcher.sendFetches();

    // 用 NetworkClient 向服务端发送拉取请求
    client.poll(pollTimeout, startMs, () -> return !fetcher.hasCompletedFetches());

    return fetcher.fetchedRecords(); // 再次从缓存队列拉取
}
// 从缓存拉取队列拉取消息
public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
    Map<TopicPartition, List<ConsumerRecord<K, V>>> fetched = new HashMap<>();
    int recordsRemaining = maxPollRecords;

    while (recordsRemaining > 0) { // 在超时时间内不断轮询
        if (nextInLineRecords == null || nextInLineRecords.isFetched) { // 分区记录为空，或者已拉取
            CompletedFetch completedFetch = completedFetches.peek(); // 从缓存队列拉取消息
            nextInLineRecords = parseCompletedFetch(completedFetch); // 将消息解析成分区消息记录 PartitionRecords
            completedFetches.poll(); // 对缓存队列移除
        } else {
            List<ConsumerRecord<K, V>> records = fetchRecords(nextInLineRecords, recordsRemaining); // 从分区记录拉取消息
            TopicPartition partition = nextInLineRecords.partition;
            if (!records.isEmpty()) { // 拉取到消息，方法 Map，以返回给消费者
                fetched.put(partition, records);
        }
    }
    return fetched;
}

4 总体流程