KafkaProducer网络模型分析

/**
 * 创建 node 的连接, 如果已经连接, 则返回 true; 否则创建一个连接
 */
public boolean ready(Node node, long now) {
	if (node.isEmpty())
		throw new IllegalArgumentException("Cannot connect to empty node " + node);

	if (isReady(node, now))
		return true;

	if (connectionStates.canConnect(node.idString(), now))
		// 如果没有对应 node 的连接, 则创建一个
		initiateConnect(node, now);

	return false;
}

/**
 * 将 Request 放入缓冲区, 等待发送。请求对应的节点 ready 才能被发送 
 */
public void send(ClientRequest request, long now) {
	doSend(request, false, now);
}

private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) {
	String nodeId = clientRequest.destination();
	if (!isInternalRequest) {
		// 如果是 NetworkClient 外部发来的请求, 则校验对应的 node是否 ready。内部的则不用交验，内部的认为是已经校验过了
		if (!canSendRequest(nodeId, now))
			throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
	}
	doSend(clientRequest, isInternalRequest, now, builder.build(version));
}

private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {
	String destination = clientRequest.destination();
	RequestHeader header = clientRequest.makeHeader(request.version());
    // 构建 send 对象
	Send send = request.toSend(destination, header);
	InFlightRequest inFlightRequest = new InFlightRequest(clientRequest, header, isInternalRequest,
			request, send, now);
	/** 先加入队列, 还没开始发送 */
	this.inFlightRequests.add(inFlightRequest);
	// 暂存至 KafkaChannel 中的 Send 对象中
	selector.send(send);
}

/**
 * 执行网络请求
 */
@Override
public List<ClientResponse> poll(long timeout, long now) {
	// 处理在发送过程中出现 UnsupportedVersionException 的 request
	if (!abortedSends.isEmpty()) {
		// 不需要进行本次 poll() 操作, 直接进行快速失败
		List<ClientResponse> responses = new ArrayList<>();
		handleAbortedSends(responses);
		completeResponses(responses);
		return responses;
	}
	/** 如有需要, 则更新Metadata */
	long metadataTimeout = metadataUpdater.maybeUpdate(now);
	try {
		/** 执行网络I/O操作 */
		this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
	} catch (IOException e) {
		log.error("Unexpected error during I/O", e);
	}

	// 处理完成后的操作
	long updatedNow = this.time.milliseconds();
	List<ClientResponse> responses = new ArrayList<>();
	/** 完成发送的处理器, 处理 completedSends 队列 */
	handleCompletedSends(responses, updatedNow);
	/** 完成接收的处理器, 处理 completedReceives 队列 */
	handleCompletedReceives(responses, updatedNow);
	/** 断开连接的处理器, 处理 disconnected 列表 */
	handleDisconnections(responses, updatedNow);
	/** 处理连接的处理器, 处理connected 列表 */
	handleConnections();
	handleInitiateApiVersionRequests(updatedNow);
	/** 超时请求的处理器, 处理 InFlightRequests 中超时的请求 */
	handleTimedOutRequests(responses, updatedNow);
	// 执行所有 responses 中的回调
	completeResponses(responses);

	return responses;
}

/**
 * 创建Server端连接, 并关联到 nioSelector
 * 注意：这里面初始化的连接是异步的，调用poll()的时候, 需要调用 connected() 去检查是否连接成功
 */
public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
    // 创建SocketChannel
	SocketChannel socketChannel = SocketChannel.open();
	SelectionKey key = null;
	try {
	    // 配置 SocketChannel为非阻塞等
		configureSocketChannel(socketChannel, sendBufferSize, receiveBufferSize);
		/** 因为非阻塞模式, 可能在连接建立之前就已经返回, 后面会通过 finishConnect() 确认是否真正建立 */
		boolean connected = doConnect(socketChannel, address);
		// 在nioSelector 上面注册 OP_CONNECT 事件
		key = registerChannel(id, socketChannel, SelectionKey.OP_CONNECT);

		if (connected) {
			// OP_CONNECT won't trigger for immediately connected channels
			log.debug("Immediately connected to node {}", id);
			immediatelyConnectedKeys.add(key);
			key.interestOps(0);
		}
	} catch (IOException | RuntimeException e) {
		if (key != null)
			immediatelyConnectedKeys.remove(key);
		channels.remove(id);
		socketChannel.close();
		throw e;
	}
}

protected SelectionKey registerChannel(String id, SocketChannel socketChannel, int interestedOps) throws IOException {
	SelectionKey key = socketChannel.register(nioSelector, interestedOps);
	KafkaChannel channel = buildAndAttachKafkaChannel(socketChannel, id, key);
	// 将 nodeId 和 kafkaChannel 绑定, 放到 channels 里面管理
	this.channels.put(id, channel);
	if (idleExpiryManager != null)
		idleExpiryManager.update(channel.id(), time.nanoseconds());
	return key;
}

/**
 * 将 send 缓存到 KafkaChannel 的 send 字段中
 */
public void send(Send send) {
	String connectionId = send.destination();
	KafkaChannel channel = openOrClosingChannelOrFail(connectionId);
	if (closingChannels.containsKey(connectionId)) {
		// 如果连接关闭, 则加入到失败集合中
		this.failedSends.add(connectionId);
	} else {
		try {
			channel.setSend(send);
		} catch (Exception e) {
			channel.state(ChannelState.FAILED_SEND);
		}
	}
}

/**
 * Selector 轮询事件, 分别调用 KafkaChannel 的 read() 和 write()
 */
public void poll(long timeout) throws IOException {
	boolean madeReadProgressLastCall = madeReadProgressLastPoll;
	/** 清除上一次 poll 结果 */
	clear();

	boolean dataInBuffers = !keysWithBufferedRead.isEmpty();

	if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty() || (madeReadProgressLastCall && dataInBuffers))
		timeout = 0;

	/* check ready keys */
	long startSelect = time.nanoseconds();
	/** 立即进行一次轮询, 或者阻塞(直到超时) */
	int numReadyKeys = select(timeout);
	long endSelect = time.nanoseconds();
	this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

	if (numReadyKeys > 0 || !immediatelyConnectedKeys.isEmpty() || dataInBuffers) {
		Set<SelectionKey> readyKeys = this.nioSelector.selectedKeys();

		// Poll from channels that have buffered data (but nothing more from the underlying socket)
		if (dataInBuffers) {
			keysWithBufferedRead.removeAll(readyKeys); //so no channel gets polled twice
			Set<SelectionKey> toPoll = keysWithBufferedRead;
			keysWithBufferedRead = new HashSet<>(); //poll() calls will repopulate if needed
			pollSelectionKeys(toPoll, false, endSelect);
		}

		/** 处理 I/O 事件的核心 */
		pollSelectionKeys(readyKeys, false, endSelect);
		// Clear all selected keys so that they are included in the ready count for the next select
		readyKeys.clear();

		pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
		immediatelyConnectedKeys.clear();
	} else {
		madeReadProgressLastPoll = true; //no work is also "progress"
	}
}