聊聊 Kafka： Consumer 源码解析之 ConsumerNetworkClient

/**
 * @author: 微信公众号【老周聊架构】
 */
public class KafkaConsumerTest {
    public static void main(String[] args) {
        Properties props = new Properties();
        // kafka地址,列表格式为host1:port1,host2:port2,...，无需添加所有的集群地址，kafka会根据提供的地址发现其他的地址(建议多提供几个，以防提供的服务器关闭) 必须设置
        props.put("bootstrap.servers", "localhost:9092");
        // key序列化方式 必须设置
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化方式 必须设置
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("group.id", "consumer_riemann_test");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 可消费多个topic,组成一个list
        String topic = "riemann_kafka_test";
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

long trySend(long now) {
    long pollDelayMs = maxPollTimeoutMs;
    // send any requests that can be sent now
    // 遍历 unsent 集合
    for (Node node : unsent.nodes()) {
        Iterator<ClientRequest> iterator = unsent.requestIterator(node);
        if (iterator.hasNext())
            pollDelayMs = Math.min(pollDelayMs, client.pollDelayMs(node, now));
        while (iterator.hasNext()) {
            ClientRequest request = iterator.next();
            // 调用 NetworkClient.ready(）检查是否可以发送请求
            if (client.ready(node, now)) {
                // 调用 NetworkClient.send(）方法，等待发送请求。
                client.send(request, now);
                // 从 unsent 集合中删除此请求
                iterator.remove();
            } else {
                // try next node when current node is not ready
                break;
            }
        }
    }
    return pollDelayMs;
}

private void checkDisconnects(long now) {
    // any disconnects affecting requests that have already been transmitted will be handled
    // by NetworkClient, so we just need to check whether connections for any of the unsent
    // requests have been disconnected; if they have, then we complete the corresponding future
    // and set the disconnect flag in the ClientResponse
    for (Node node : unsent.nodes()) {
        // 检测消费者与每个 Node 之间的连接状态
        if (client.connectionFailed(node)) {
            // Remove entry before invoking request callback to avoid callbacks handling
            // coordinator failures traversing the unsent list again.
            // 在调用请求回调之前删除条目以避免回调处理再次遍历未发送列表的协调器故障。
            Collection<ClientRequest> requests = unsent.remove(node);
            for (ClientRequest request : requests) {
                RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
                AuthenticationException authenticationException = client.authenticationException(node);
                // 调用 ClientRequest 的回调函数
                handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().latestAllowedVersion()),
                        request.callback(), request.destination(), request.createdTimeMs(), now, true,
                        null, authenticationException, null));
            }
        }
    }
}

public void maybeTriggerWakeup() {
    // 通过 wakeupDisabled 检测是否在执行不可中断的方法，通过 wakeup 检测是否有中断请求。
    if (!wakeupDisabled.get() && wakeup.get()) {
        log.debug("Raising WakeupException in response to user wakeup");
        // 重置中断标志
        wakeup.set(false);
        throw new WakeupException();
    }
}

private void failExpiredRequests(long now) {
    // clear all expired unsent requests and fail their corresponding futures
    // 清除所有过期的未发送请求并使其相应的 futures 失败
    Collection<ClientRequest> expiredRequests = unsent.removeExpiredRequests(now);
    for (ClientRequest request : expiredRequests) {
        RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
        // 调用回调函数
        handler.onFailure(new TimeoutException("Failed to send request after " + request.requestTimeoutMs() + " ms."));
    }
}
private Collection<ClientRequest> removeExpiredRequests(long now) {
    List<ClientRequest> expiredRequests = new ArrayList<>();
    for (ConcurrentLinkedQueue<ClientRequest> requests : unsent.values()) {
        Iterator<ClientRequest> requestIterator = requests.iterator();
        while (requestIterator.hasNext()) {
            ClientRequest request = requestIterator.next();
            // 检查是否超时
            long elapsedMs = Math.max(0, now - request.createdTimeMs());
            if (elapsedMs > request.requestTimeoutMs()) {
                // 将过期请求加入到 expiredRequests 集合
                expiredRequests.add(request);
                requestIterator.remove();
            } else
                break;
        }
    }
    return expiredRequests;
}

public RequestFuture<ClientResponse> send(Node node,
                                          AbstractRequest.Builder<?> requestBuilder,
                                          int requestTimeoutMs) {
    long now = time.milliseconds();
    RequestFutureCompletionHandler completionHandler = new RequestFutureCompletionHandler();
    ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, true,
            requestTimeoutMs, completionHandler);
    // 创建 clientRequest 对象，并保存到 unsent 集合中。
    unsent.put(node, clientRequest);
    // wakeup the client in case it is blocking in poll so that we can send the queued request
    // 唤醒客户端以防它在轮询中阻塞，以便我们可以发送排队的请求。
    client.wakeup();
    return completionHandler.future;
}

/**
 * 适配器
 * Adapt from a request future of one type to another.
 *
 * @param <F> Type to adapt from
 * @param <T> Type to adapt to
 */
public abstract class RequestFutureAdapter<F, T> {
    public abstract void onSuccess(F value, RequestFuture<T> future);
    public void onFailure(RuntimeException e, RequestFuture<T> future) {
        future.raise(e);
    }
}
/**
 * RequestFuture<T> 适配成 RequestFuture<S>
 * Convert from a request future of one type to another type
 * @param adapter The adapter which does the conversion
 * @param <S> The type of the future adapted to
 * @return The new future
 */
public <S> RequestFuture<S> compose(final RequestFutureAdapter<T, S> adapter) {
    // 适配之后的结果
    final RequestFuture<S> adapted = new RequestFuture<>();
    // 在当前 RequestFuture 上添加监听器
    addListener(new RequestFutureListener<T>() {
        @Override
        public void onSuccess(T value) {
            adapter.onSuccess(value, adapted);
        }
        @Override
        public void onFailure(RuntimeException e) {
            adapter.onFailure(e, adapted);
        }
    });
    return adapted;
}

public void chain(final RequestFuture<T> future) {
    // 添加监听器
    addListener(new RequestFutureListener<T>() {
        @Override
        public void onSuccess(T value) {
            // 通过监听器将 value 传递给下一个 RequestFuture 对象
            future.complete(value);
        }
        @Override
        public void onFailure(RuntimeException e) {
            // 通过监听器将异常传递给下一个 RequestFuture 对象
            future.raise(e);
        }
    });
}