KafkaConsumer网络模型分析

public class CustomConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 定义kakfa 服务的地址，不需要将所有broker指定上
        props.put("bootstrap.servers", "127.0.0.1:9092");
        // 制定consumer group
        props.put("group.id", "g1");
        // 是否自动确认offset
        props.put("enable.auto.commit", "true");
        // 自动确认offset的时间间隔
        props.put("auto.commit.interval.ms", "1000");
        // key的序列化类
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // value的序列化类
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 定义consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 消费者订阅的topic, 可同时订阅多个
        consumer.subscribe(Arrays.asList("test1"));
        while (true) {
            // 读取数据，读取超时时间为100ms
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

/**
 * 从服务端获取消息
 */
public ConsumerRecords<K, V> poll(final Duration timeout) {
	return poll(time.timer(timeout), true);
}

private ConsumerRecords<K, V> poll(final Timer timer, final boolean includeMetadataInTimeout) {
    // 确保消费者未关闭
	acquireAndEnsureOpen();
	try {
		if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) {
			throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
		}

		// 拉取消息直到超时
		do {
			client.maybeTriggerWakeup();

			if (includeMetadataInTimeout) {
				if (!updateAssignmentMetadataIfNeeded(timer)) {
					return ConsumerRecords.empty();
				}
			} else {
				/** 循环直到更新 metadata */
				while (!updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE))) {
					log.warn("Still waiting for metadata");
				}
			}
			/** 客户端拉取消息的核心逻辑 */
			final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(timer);
			if (!records.isEmpty()) {
				/** 在返回数据之前, 发送下次的 fetch 请求, 避免用户在下次获取数据时线程阻塞 */
				if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {
				    // 如果sendFetches() 数据到缓冲区或有unsent有数据, 则执行poll() 执行网络IO
					client.pollNoWakeup();
				}
				return this.interceptors.onConsume(new ConsumerRecords<>(records));
			}
		} while (timer.notExpired());

		return ConsumerRecords.empty();
	} finally {
		release();
	}
}

/** 一次拉取过程, 除了获取新数据外, 还会做一些必要的 offset commit 和 offset reset 的操作 */
private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(Timer timer) {
	long pollTimeout = coordinator == null ? timer.remainingMs() : Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs());

	// 如果数据已经获取到了, 则立即返回
	final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();
	if (!records.isEmpty()) {
		return records;
	}

	/** 外层循环会不断循环调用 pollForFetches(), 在返回给调用者之前, 会再次调用 sendFetches() + client.poll() 发起第二个网络请求 */
	fetcher.sendFetches();
	if (!cachedSubscriptionHashAllFetchPositions && pollTimeout > retryBackoffMs) {
		pollTimeout = retryBackoffMs;
	}
	/** 执行真正的网络请求 */
	Timer pollTimer = time.timer(pollTimeout);
	client.poll(pollTimer, () -> {
		/** 有完成的 fetcher 请求的话, 这里就不会阻塞; 阻塞的话也有超时时间 */
		return !fetcher.hasCompletedFetches();
	});
	timer.update(pollTimer.currentTimeMs());

	/** 如果 group 需要 rebalance, 直接返回空数据, 这样可以更快的让 group 进入稳定状态 */
	if (coordinator != null && coordinator.rejoinNeededOrPending()) {
		return Collections.emptyMap();
	}

	return fetcher.fetchedRecords();
}

/**
 * 如果一个 node 已经分配分区, 并且没有处理中的 Fetch 数据, 此时可以创建 FetchRequest 发送
 * @return 发送了多少个 fetches 请求
 */
public synchronized int sendFetches() {
	// Update metrics in case there was an assignment change
	sensors.maybeUpdateAssignment(subscriptions);
    // 按照 Node 的维度，构造 FetchRequest 请求
	Map<Node, FetchSessionHandler.FetchRequestData> fetchRequestMap = prepareFetchRequests();
	for (Map.Entry<Node, FetchSessionHandler.FetchRequestData> entry : fetchRequestMap.entrySet()) {
		final Node fetchTarget = entry.getKey();
		final FetchSessionHandler.FetchRequestData data = entry.getValue();
		// 构建请求
		final FetchRequest.Builder request = FetchRequest.Builder.forConsumer(this.maxWaitMs, this.minBytes, data.toSend())
				.isolationLevel(isolationLevel).setMaxBytes(this.maxBytes).metadata(data.metadata()).toForget(data.toForget()).rackId(clientRackId);

		if (log.isDebugEnabled()) {
			log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget);
		}
		// 发送请求
		client.send(fetchTarget, request).addListener(new RequestFutureListener<ClientResponse>() {
			@Override
			public void onSuccess(ClientResponse resp) {
				synchronized (Fetcher.this) {
					try {
						@SuppressWarnings("unchecked")
						FetchResponse<Records> response = (FetchResponse<Records>) resp.responseBody();
						FetchSessionHandler handler = sessionHandler(fetchTarget.id());
						if (handler == null) {
							log.error("Unable to find FetchSessionHandler for node {}. Ignoring fetch response.", fetchTarget.id());
							return;
						}
						if (!handler.handleResponse(response)) {
							return;
						}

						Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
						FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
						/** 遍历所有响应中的数据 */
						for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
							TopicPartition partition = entry.getKey();
							FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
							if (requestData == null) {
								String message;
								if (data.metadata().isFull()) {
									message = MessageFormatter.arrayFormat("Response for missing full request partition: partition={}; metadata={}", new Object[]{partition, data.metadata()}).getMessage();
								} else {
									message = MessageFormatter.arrayFormat("Response for missing session request partition: partition={}; metadata={}; toSend={}; toForget={}", new Object[]{partition, data.metadata(), data.toSend(), data.toForget()}).getMessage();
								}

								// Received fetch response for missing session partition
								throw new IllegalStateException(message);
							} else {
								long fetchOffset = requestData.fetchOffset;
								FetchResponse.PartitionData<Records> fetchData = entry.getValue();
								/** 创建 CompletedFetch, 并缓存到completedFetches 队列中 */
								log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", isolationLevel, fetchOffset, partition, fetchData);
								completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator, resp.requestHeader().apiVersion()));
							}
						}

						sensors.fetchLatency.record(resp.requestLatencyMs());
					} finally {
					    // 从发送中队列中移除
						nodesWithPendingFetchRequests.remove(fetchTarget.id());
					}
				}
			}
			@Override
			public void onFailure(RuntimeException e) {
				synchronized (Fetcher.this) {
					try {
						FetchSessionHandler handler = sessionHandler(fetchTarget.id());
						if (handler != null) {
							handler.handleError(e);
						}
					} finally {
					    // 从发送中队列中移除
						nodesWithPendingFetchRequests.remove(fetchTarget.id());
					}
				}
			}
		});
    /** 加入发送中队列 */
		this.nodesWithPendingFetchRequests.add(entry.getKey().id());
	}
	return fetchRequestMap.size();
}

/**
 * 拉取数据（poll()）完成后, 存储在 completedFetches 缓存中的数据尚未解析. 此时调用 fetchedRecords() 解析并返回
 * @return 按照分区维度的消息记录
 */
public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
	Map<TopicPartition, List<ConsumerRecord<K, V>>> fetched = new HashMap<>();
	int recordsRemaining = maxPollRecords;

	try {
		while (recordsRemaining > 0) {
			if (nextInLineRecords == null || nextInLineRecords.isFetched) {
				/** 上一个分区缓存中数据已处理完，则从分区队列中获取下一个分区缓存数据 */
				CompletedFetch completedFetch = completedFetches.peek();
				if (completedFetch == null) break;

				try {
					/** 解析分区缓存数据 CompletedFetch, 得到一个 PartitionRecords */
					nextInLineRecords = parseCompletedFetch(completedFetch);
				} catch (Exception e) {
					FetchResponse.PartitionData partition = completedFetch.partitionData;
					if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) {
						completedFetches.poll();
					}
					throw e;
				}
				completedFetches.poll();
			} else {
			    // 从分区缓存中获取指定条数的消息
				List<ConsumerRecord<K, V>> records = fetchRecords(nextInLineRecords, recordsRemaining);
				TopicPartition partition = nextInLineRecords.partition;
				if (!records.isEmpty()) {
					List<ConsumerRecord<K, V>> currentRecords = fetched.get(partition);
					if (currentRecords == null) {
						fetched.put(partition, records);
					} else {
						// this case shouldn't usually happen because we only send one fetch at a time per partition,
						// but it might conceivably happen in some rare cases (such as partition leader changes).
						// we have to copy to a new list because the old one may be immutable
						List<ConsumerRecord<K, V>> newRecords = new ArrayList<>(records.size() + currentRecords.size());
						newRecords.addAll(currentRecords);
						newRecords.addAll(records);
						fetched.put(partition, newRecords);
					}
					recordsRemaining -= records.size();
				}
			}
		}
	} catch (KafkaException e) {
		if (fetched.isEmpty())
			throw e;
	}
	return fetched;
}