KafkaConsumer 组件源码 ConsumerNetworkClient

负责缓存要发送的请求、将加了回调的请求交给NetworkClient、触发NetworkClient的IO。

主要功能

主要方法是send和poll。调用send将请求缓存，调用poll让NetworkClient执行IO，其中trySend将加了回调逻辑的请求交给NetworkClient。

回调实现

ConsumerNetworkClient的请求-响应处理是由自定义的RequestFutureCompletionHandler和RequestFuture协同实现的。RequestFutureCompletionHandler会包含并暴露一个RequestFuture，供外界调用。

public RequestFuture<ClientResponse> send(Node node, AbstractRequest.Builder<?> requestBuilder) {
        long now = time.milliseconds();
        RequestFutureCompletionHandler completionHandler = new RequestFutureCompletionHandler();
        ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, true,
                completionHandler);
        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;
    }

RequestFutureCompletionHandler继承了RequestCompletionHandler，因此作为回调传入client.newClientRequest，并实现了自己的回调逻辑，在被回调时触发RequestFuture的完成。

private class RequestFutureCompletionHandler implements RequestCompletionHandler {

在RequestFuture中可以添加监听器，当请求完成时进行回调。

两者的行为逻辑如下:

请求缓存

ConsumerNetworkClient利用unsent存储尚未发出的请求。

用ConsumerNetworkClient封装的意义

目前想到一个意义，加快客户端调用:

• 原本没进行封装时，NetworkClient在send前需要先判断ready再send
• 经过ConsumerNetworkClient的包装，直接调用send即可，要发送的消息会存在unsent变量中，调用poll送入网络。这样send会将请求缓存到unset后，快速返回，不耽误调用者的时间。