KafkaController分析2-NetworkClient分析InFlightRequests类

  • NetworkClient: 顾名思义哈, 用于网络连接,消息发送的客户端封装, 源码中的注释:

A network client for asynchronous request/response network i/o. This is an internal class used to implement the user-facing producer and consumer clients.

  • 用在何处:
1. kafka本身实现了java版的producer和consumer,里面的网络连接,请求发送均使用NetworkClient实现;
2. KafkaController中controller与其他broker的通讯,使用NetworkClient实现;

InFlightRequests类

  • 所在文件: clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
  • 实现了request的集合, 包括正在发送的和已经发送的但还没有接收到response的request;
  • 主要成员变量: private final Map<String, Deque<ClientRequest>> requests = new HashMap<String, Deque<ClientRequest>>(); 针对每个连接使用Deque<ClientRequest>数据结构来保存所有的request;Deque<ClientRequest> 是个双端队列;
  • 添加新的request, 新的reqeust总是通过addFirst放到队首
public void add(ClientRequest request) {
        Deque<ClientRequest> reqs = this.requests.get(request.request().destination());
        if (reqs == null) {
            reqs = new ArrayDeque<>();
            this.requests.put(request.request().destination(), reqs);
        }
        reqs.addFirst(request);
    }
  • 取出最早发送的request, 通过pollLast()取出
public ClientRequest completeNext(String node) {
        return requestQueue(node).pollLast();
    }
  • public boolean canSendMore(String node)决定是否可以通过NetworkClient来发送请求 对于通过NetworkClient来发送的request, 如果之前发送的请求并没有通过底层socket实际发送完成, 是不允许发送新的request的
public boolean canSendMore(String node) {
        Deque<ClientRequest> queue = requests.get(node);
        return queue == null || queue.isEmpty() ||
               (queue.peekFirst().request().completed() && queue.size() < this.maxInFlightRequestsPerConnection);
    }
```

# ClusterConnectionStates
* 所在文件:clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
* 记录到各个broker node的连接状态:
`private final Map<String, NodeConnectionState> nodeState`
* 对同一node的两次连接有一定的时间间隔限制, 即采用延迟连接:
`private final long reconnectBackoffMs`
* 连接状态有如下三种:
```
ConnectionState.DISCONNECTED -- 未连接
ConnectionState.DISCONNECTING -- 正在连接
ConnectionState.CONNECTED -- 已连接
```
* `canConnect`: 判断是否允许连接到node:如果从未连接过或者连接当前是断开的并且距离上次连接的间隔大于`reconnectBackoffMs`, 则允许连接;
```
public boolean canConnect(String id, long now) {
        NodeConnectionState state = nodeState.get(id);
        if (state == null)
            return true;
        else
            return state.state == ConnectionState.DISCONNECTED && now - state.lastConnectAttemptMs >= this.reconnectBackoffMs;
    }
```

# NetworkClien类
* 所在文件: clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
* 非线程安全
* 继承自 `KafkaClient`
* 使用了 `org.apache.kafka.common.network.Selector`来处理网络IO, [详情点这里 => Kafka源码分析-网络层](http://www.jianshu.com/p/8cbc7618abcb)
* 简单讲这个类用来管理一个到broker node的连接,请求发送和响应接收:
>A network client for asynchronous request/response network i/o. This is an internal class used to implement the user-facing producer and consumer clients.
* 核心函数 `poll`
使用`selector.poll`来处理实现的socket读写事件;
```
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }
```
经过`selector.poll`的调用,所有**发送完成的requet**, **接收完成的response**, **所有断开的连接**, **所有新建成功的连接**都将放到`selector`中相应的队列里;
* 处理发送完成的request
```
private void handleCompletedSends(List<ClientResponse> responses, long now) {
        // if no response is expected then when the send is completed, return it
        for (Send send : this.selector.completedSends()) {
            ClientRequest request = this.inFlightRequests.lastSent(send.destination());
            if (!request.expectResponse()) {
                this.inFlightRequests.completeLastSent(send.destination());
                responses.add(new ClientResponse(request, now, false, null));
            }
        }
    }
```
对于不需要回应response的请求,将从`ifFlightRequests`中删除;
* 处理接收到的response
```
private void handleCompletedReceives(List<ClientResponse> responses, long now) {
        for (NetworkReceive receive : this.selector.completedReceives()) {
            String source = receive.source();
            ClientRequest req = inFlightRequests.completeNext(source);
            ResponseHeader header = ResponseHeader.parse(receive.payload());
            // Always expect the response version id to be the same as the request version id
            short apiKey = req.request().header().apiKey();
            short apiVer = req.request().header().apiVersion();
            Struct body = ProtoUtils.responseSchema(apiKey, apiVer).read(receive.payload());
            correlate(req.request().header(), header);
            if (!metadataUpdater.maybeHandleCompletedReceive(req, now, body))
                responses.add(new ClientResponse(req, now, false, body));
        }
    }
```
如果是`metadata`的更新response,则调用`metadataUpdater.maybeHandleCompletedReceive` 处理metadata的更新;
* 处理新建的连接
```
 private void handleConnections() {
        for (String node : this.selector.connected()) {
            log.debug("Completed connection to node {}", node);
            this.connectionStates.connected(node);
        }
    }
```
* 处理所有的`handle***`函数返回的responses
```
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleTimedOutRequests(responses, updatedNow);

        // invoke callbacks
        for (ClientResponse response : responses) {
            if (response.request().hasCallback()) {
                try {
                    response.request().callback().onComplete(response);
                } catch (Exception e) {
                    log.error("Uncaught error in request completion:", e);
                }
            }
        }
```

# NetworkClientBlockingOps
* 所在文件: core/src/main/scala/kafka/utils/NetworkClientBlockingOps.scala
* 利用非阻塞的`NetworkClient`的方法, 实现了阻塞的方法;
* 阻塞直到`Client.ready`
```
def blockingReady(node: Node, timeout: Long)(implicit time: JTime): Boolean = {
    client.ready(node, time.milliseconds()) || pollUntil(timeout) { (_, now) =>
      if (client.isReady(node, now))
        true
      else if (client.connectionFailed(node))
        throw new IOException(s"Connection to $node failed")
      else false
    }
  }
```
* 阻塞发送request直到收到response
```
def blockingSendAndReceive(request: ClientRequest, timeout: Long)(implicit time: JTime): Option[ClientResponse] = {
    client.send(request, time.milliseconds())

    pollUntilFound(timeout) { case (responses, _) =>
      val response = responses.find { response =>
        response.request.request.header.correlationId == request.request.header.correlationId
      }
      response.foreach { r =>
        if (r.wasDisconnected) {
          val destination = request.request.destination
          throw new IOException(s"Connection to $destination was disconnected before the response was read")
        }
      }
      response
    }
  }
```

##### [Kafka源码分析-汇总](http://www.jianshu.com/p/aa274f8fe00f)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Android 研究

Android Handler机制11之Handler机制总结

经过上面的思考,大家是不是发现和其实我们Handler的机制基本上一致。Looper负责轮询;Message代表消息,为了区别对待,用what来做为标识符,wh...

1K1
来自专栏Coding迪斯尼

从0到1用java再造tcpip协议栈:实现ARP协议层

经过前两节的准备,我们完成了数据链路层,已经具备了数据包接收和发送的基础设施,本机我们在此基础上实现上层协议,我们首先从实现ARP协议开始。先简单认识一下ARP...

1972
来自专栏恰童鞋骚年

自己动手写工具:百度图片批量下载器

开篇:在某些场景下,我们想要对百度图片搜出来的东东进行保存,但是一个一个得下载保存不仅耗时而且费劲,有木有一种方法能够简化我们的工作量呢,让我们在离线模式下也能...

4061
来自专栏移动开发的那些事儿

BlockCanary源码解析

如上代码中的loop()方法是Looper中的,我们的目的是监测主线程的卡顿问题,因为UI更新界面都是在主线程中进行的,所以在主线程中做耗时操作可能会造成界面卡...

1412
来自专栏SDNLAB

源码解读ODL的MAC地址学习(二)

1 简介 上一篇文章(源码解读ODL的MAC地址学习(一))已经分析了MAC地址学习中的ARP请求的部分源码,下面将接着上一篇文章,介绍一下ARP响应和生成流表...

4315
来自专栏林冠宏的技术文章

XGoServer 一个基础性、模块完整且安全可靠的服务端框架

作者:林冠宏 / 指尖下的幽灵 掘金:https://juejin.im/user/587f0dfe128fe100570ce2d8 博客:htt...

39218
来自专栏coder修行路

Go 源码学习之--net/http

其实自己不是很会看源码,但是学习优秀的源码是提升自己代码能力的一种方式,也可以对自己以后写代码有一个很好的影响,所以决定在之后的时间内,要有一个很好的习惯,阅读...

7305
来自专栏李蔚蓬的专栏

1.Android系统源代码目录与系统目录

想要看完整个Android的源代码,需要懂C、懂脚本、懂Java、软硬兼通。所以一般情况下,我们了解源代码的框架结构,出了问题知道从哪里着手解决就可以了。这就好...

5672
来自专栏前端黑板报

HTTP2基础教程-读书笔记(四)

? 记录一下HTTP/2的底层原理,帮助理解协议实现细节。 连接 每个端点都需要发送一个连接作为最终确认使用的协议,并建立http/2连接的初始设置。客户端和...

3516
来自专栏数值分析与有限元编程

Fortran知识 | 输出结果出现NaN

一旦输出结果出现NaN,编译器不会给出任何错误提示,这个时候该如何调试程序呢? ? 点击菜单栏的调试,最后一个为XXXX属性,打开对话框,左侧展开 Fortra...

3747

扫码关注云+社区

领取腾讯云代金券