1、一个batch什么条件下可以发送出去

• 上面我们介绍了Sender线程发送那个消息的大概流程，接下来我们来分析一下一个batch的数据在什么情况下会发送出去？

• 回顾发送消息的时候，生产者需要指定的相关参数

  retries   : 重试的次数,默认为0
  linger.ms : 生产者在发送批次之前等待更多消息加入批次的时间，默认为0，表示不等待
  retry.backoff.ms:重试的时间间隔，默认100ms
  

• 核心代码Sender类当中的run方法当中定义以下代码

  /***
   *  步骤二：
   *   判断哪些partition有消息可以发送，获取到这个partition的leaderPartition对应的broker主机
   *
   */
  // get the list of partitions with data ready to send
  RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
  

• accumulator.ready方法剖析

   public ReadyCheckResult ready(Cluster cluster, long nowMs) {
          Set<Node> readyNodes = new HashSet<>();
          long nextReadyCheckDelayMs = Long.MAX_VALUE;
          Set<String> unknownLeaderTopics = new HashSet<>();
  
          //todo: waiters队列里面有数据，说明内存池中的内存不够了。
          //如果exhausted为true，说明内存池中的内存不够了
          boolean exhausted = this.free.queued() > 0;
  
          //todo: 遍历所有的分区，一个分区对应一个队列
          for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {
              //获取分区
              TopicPartition part = entry.getKey();
              //获取分区对应的队列
              Deque<RecordBatch> deque = entry.getValue();
  
              //todo: 根据分区获取该分区的leader partition对应的主机
              Node leader = cluster.leaderFor(part);
              synchronized (deque) {
                  //todo: 如果没有找到对应的主机， 进行标识，下一次重新获取该topic的元数据信息
                  if (leader == null && !deque.isEmpty()) {
                      // This is a partition for which leader is not known, but messages are available to send.
                      // Note that entries are currently not removed from batches when deque is empty.
                      unknownLeaderTopics.add(part.topic());
  
                  } else if (!readyNodes.contains(leader) && !muted.contains(part)) {
                      //todo: 获取队列中第一个批次
                      RecordBatch batch = deque.peekFirst();
  
                      //todo: 批次不为null, 判断该批次是否可以发送
                      if (batch != null) {
                          /**
                           * batch.attempts:重试的次数
                           * batch.lastAttemptMs ： 上一次重试的时间
                           * retryBackoffMs ： 重试的时间间隔
                           *
                           * backingOff ：表示重新发送数据的时间到了。
                           */
                          boolean backingOff = batch.attempts > 0 && batch.lastAttemptMs + retryBackoffMs > nowMs;
  
                          /**
                           * nowMs :当前时间
                           * batch.lastAttemptMs ： 上一次重试的时间
                           * waitedTimeMs ：该批次已经等待了多久
                           */
                          long waitedTimeMs = nowMs - batch.lastAttemptMs;
  
                          /**
                           *  使用场景驱动的方式去分析，第一次发送数据，之前也没有消息发送过，也没有重试这一说
                           *  backingOff 应该为false
                           *
                           *  timeToWaitMs=lingerMs 默认就是0 ，表示不需要等待
                           *  如果默认是0的话，就是来一条消息就发送一条信息，很明显是不合适的，
                           *  所以我们在发送数据的时候，一定要记得去配置这个参数
                           *  假设我们配置的是100ms,表示消息最多等待100ms后，必须要发送出去
                           */
                          long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
  
                          /**
                           * timeToWaitMs :最多能等多久
                           * waitedTimeMs ：已经等待了多久
                           * timeLeftMs：还能等待多久
                           */
                          long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
  
                          /**
                           *  deque.size()： 队列中至少有2个批次，这里至少会出现1个批次是写满了
                           *  batch.records.isFull()：该批次已经写满了
                           *
                           *  有可能出现只有1个批次，该批次数据已经写满了，也需要把数据发送出去
                           *  full： 是否有写满的批次
                           */
                          boolean full = deque.size() > 1 || batch.records.isFull();
  
                          /**
                           * waitedTimeMs: 已经等待的时间
                           * timeToWaitMs：最多能等多久
                           *
                           * expired：是否达到了最大的等待时间
                           */
                          boolean expired = waitedTimeMs >= timeToWaitMs;
  
                          /**
                           * （1）full:    如果一个批次写满了，无论时间有没有到，都需要把消息发送出去
                           * （2）expired：时间到了，不管批次数据是否写满，也需要把消息发送出去
                           * （3）exhausted ：内存不够了，也需要把消息发送出去，最后释放内存
                           *  (4)(5) closed/flushInProgress :producer客户端线程关闭，也需要把消息发送出去
                           */
                          boolean sendable = full || expired || exhausted || closed || flushInProgress();
  
                          /**
                           * 可以发送消息了
                           */
                          if (sendable && !backingOff) {
                              //todo：把可以发送该批次数据分区的leader主机添加到集合中
                              readyNodes.add(leader);
                          } else {
                              // Note that this results in a conservative estimate since an un-sendable partition may have
                              // a leader that will later be found to have sendable data. However, this is good enough
                              // since we'll just wake up and then sleep again for the remaining time.
                              nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
                          }
                      }
                  }
              }
          }
  
          return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);
      }
  

2、筛选可以发送消息的broker

• 上面我们已经知道了消息需要发送给哪些对应的broker，这些broker主机会存储在Set集合中，接下来我们来分析下producer如何筛选可以发送消息的broker

• 核心代码Sender类当中的run方法

    Iterator<Node> iter = result.readyNodes.iterator();
    long notReadyTimeout = Long.MAX_VALUE;
    while (iter.hasNext()) {
        Node node = iter.next();
        /**
         *  步骤四：
         *      检查与要发送数据的主机网络是否建立好
         */
        if (!this.client.ready(node, now)) {
            iter.remove();
            notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
        }
    }
  

• NetworkClient类当中的ready方法剖析

      public boolean ready(Node node, long now) {
          //todo: 节点为空就报异常
          if (node.isEmpty())
              throw new IllegalArgumentException("Cannot connect to empty node " + node);
  
          //todo: 判断要发送消息的主机，是否具备发送消息的条件
          if (isReady(node, now))
              return true;
  
          //todo: 第一次进来应该是并没有建立好连接，判断是否可以尝试去建立好网络
          if (connectionStates.canConnect(node.idString(), now))
              // if we are interested in sending to a node and we don't have a connection to it, initiate one
                 //todo: 初始化连接
              initiateConnect(node, now);
  
          return false;
      }
  

• NetWorkClient类当中ready方法内部的的isReady方法剖析

      public boolean isReady(Node node, long now) {
          // if we need to update our metadata now declare all requests unready to make metadata requests first
          // priority
          /**
           * 条件1：!metadataUpdater.isUpdateDue(now)
           *  要发送写数据请求的时候，不能是正在更新元数据的操作
           *
           *  条件2：canSendRequest(node.idString())
           *      对应的broker是否已经建立好连接，并且可以发送数据
           *
           */
          return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString());
      }
  
  

• NetWorkClient类当中的isReady方法内的canSendRequest`判断连接是否建立好的方法剖析

      private boolean canSendRequest(String node) {
          /**
           * connectionStates.isConnected(node) :
           *    生产者缓存了多个连接，一个broker对应一个连接，从缓存中检查是否与节点建立好连接
           * selector.isChannelReady(node)：
           *   一个selector上面绑定了多个KafkaChannel（java socketChannel）,一个kafkachannel就是一个连接
           *
           * inFlightRequests.canSendMore(node)：
           *   与"max.in.flight.requests.per.connection"参数有关系
           *   表示每个向kafka  broker 节点发送消息的连接，最多能够容忍消息发送出去而没有接受到响应的个数，默认只是5
           *   这一块如果大于1 ，可能会出现分区数据乱序。
           *
           */
          return connectionStates.isConnected(node)
                  && selector.isChannelReady(node)
                  && inFlightRequests.canSendMore(node);
      }
  

• NetWorkClient类当中的ready方法内部的canConnect判断是否可以尝试去建立好网络方法剖析

      public boolean canConnect(String id, long now) {
          //todo: 从缓存中获取当前节点的缓存，如果为null就表示从来没有连接过
          NodeConnectionState state = nodeState.get(id);
          if (state == null)
              return true;
          else
              //todo: 可以从缓存中获取到连接，但是连接的状态是DISCONNECTED，
              // 并且now - state.lastConnectAttemptMs >= this.reconnectBackoffMs;
             // 当前时间-上一次连接时间 >= 重试的时间间隔，说明可以重试连接
              return state.state == ConnectionState.DISCONNECTED && now - state.lastConnectAttemptMs >= this.reconnectBackoffMs;
      }
  

• NetWorkClient类当中的ready方法内部的initiateConnect`初始化连接的方法剖析

      /**
       * Initiate a connection to the given node
       */
      private void initiateConnect(Node node, long now) {
          //todo: 获取节点id
          String nodeConnectionId = node.idString();
          try {
              log.debug("Initiating connection to node {} at {}:{}.", node.id(), node.host(), node.port());
              //todo: 把连接id和当前时间添加到缓存中
              this.connectionStates.connecting(nodeConnectionId, now);
              //todo: 与节点建立socket网络连接
              selector.connect(nodeConnectionId,
                               new InetSocketAddress(node.host(), node.port()),
                               this.socketSendBuffer,s
                               this.socketReceiveBuffer);
          } catch (IOException e) {
              /* attempt failed, we'll try again after the backoff */
              //todo: 出现异常把缓存中的连接状态置为DISCONNECTED
              connectionStates.disconnected(nodeConnectionId, now);
              /* maybe the problem is our metadata, update it */
              //todo: 可能由于元数据导致建立网络连接失败，那么更新元数据
              metadataUpdater.requestUpdate();
              log.debug("Error connecting to node {} at {}:{}:", node.id(), node.host(), node.port(), e);
          }
      }
  

  

3、kafka网络设计

• 前面我们再分析sender线程拉取元数据的时候涉及到了网络相关内容，同样sender线程发送数据到kafka集群也需要网络，接下面我们来分析下kafka内部的网络设计。

• 比如代码Sender类当中的run方法

              /**
               *  步骤四：
               *      检查与要发送数据的主机网络是否建立好
               */
              if (!this.client.ready(node, now)) {
                  iter.remove();
                  notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
              }
  
  
  

• NetworkClient.ready方法剖析

      public boolean ready(Node node, long now) {
          //todo: 节点为空就报异常
          if (node.isEmpty())
              throw new IllegalArgumentException("Cannot connect to empty node " + node);
  
          //todo: 判断要发送消息的主机，是否具备发送消息的条件
          if (isReady(node, now))
              return true;
  
          //todo: 第一次进来应该是并没有建立好连接，判断是否可以尝试去建立好网络
          if (connectionStates.canConnect(node.idString(), now))
              // if we are interested in sending to a node and we don't have a connection to it, initiate one
              //todo: 初始化连接
              initiateConnect(node, now);
  
          return false;
      }
  

• NetworkClient.isReady方法剖析

      public boolean isReady(Node node, long now) {
          // if we need to update our metadata now declare all requests unready to make metadata requests first
          // priority
          /**
           * 条件1：!metadataUpdater.isUpdateDue(now)
           *  要发送写数据请求的时候，不能是正在更新元数据的操作
           *
           *  条件2：canSendRequest(node.idString())
           *      对应的broker是否已经建立好连接，并且可以发送数据
           *
           */
          return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString());
      }
  

• NetworkClient.canSendRequest方法剖析

  private boolean canSendRequest(String node) {
          /**
           * connectionStates.isConnected(node) :
           *    生产者缓存了多个连接，一个broker对应一个连接，从缓存中检查是否与节点建立好连接
           * selector.isChannelReady(node)：
           *   一个selector上面绑定了多个KafkaChannel（java socketChannel）,一个kafkachannel就是一个连接
           *
           * inFlightRequests.canSendMore(node)：
           *   与"max.in.flight.requests.per.connection"参数有关系
           *   表示每个向kafka  broker 节点发送消息的连接，最多能够容忍消息发送出去而没有接受到响应的个数，默认只是5
           *   这一块如果大于1 ，可能会出现分区数据乱序。
           *
           */
          return connectionStates.isConnected(node)
                  && selector.isChannelReady(node)
                  && inFlightRequests.canSendMore(node);
      }
  

• 其中selector.isChannelReady就涉及到了网络通信

  • Selector类核心参数

     /** todo: 这个Selector是kafka内部封装的一个selector
       *  它底层是基于java NIO里面的Selector去封装的
       */
    public class Selector implements Selectable {
    
        public static final long NO_IDLE_TIMEOUT_MS = -1;
        private static final Logger log = LoggerFactory.getLogger(Selector.class);
    
        //该对象就是Java NIO中的Selector
        //负责网络的建立、发送网络请求、处理网络io，是kafka网络这一块的核心组件
        private final java.nio.channels.Selector nioSelector;
        //存储brokerId与KafkaChannel之间的映射关系，KafkaChannel是基于socketChannel进行了封装
        private final Map<String, KafkaChannel> channels;
        //已经发送完的请求
        private final List<Send> completedSends;
        //已经接收并且处理完成的响应,注意，这个集合并没有包括所有已接收完成的响应，stagedReceives集合也包括了一些接收完成的响应
        private final List<NetworkReceive> completedReceives;
        //已接收完成，但还没有暴露给用户的响应，一个kafkaChannel对应一个队列，一个队列对应一个连接，一个连接会有很多个响应
        private final Map<KafkaChannel, Deque<NetworkReceive>> stagedReceives;
        //在调用SocketChannel#connect方法时立即完成的SelectionKey
        private final Set<SelectionKey> immediatelyConnectedKeys;
        //已断开连接的节点
        private final List<String> disconnected;
        //新连接成功的节点
        private final List<String> connected;
        //发送失败的节点
        private final List<String> failedSends;
    }
    

• KafkaChannel类

    //todo: 这个kafkaChannel就是对java NIO中的socketChannel进行了封装。
    public class KafkaChannel {
        //一个broker对应一个kafkaChannel，这里就是brokerId
        private final String id;
        //封装了java NIO中的socketChannel
        private final TransportLayer transportLayer;
        //kafka安全机制认证
        private final Authenticator authenticator;
        private final int maxReceiveSize;
        //接受到的响应
        private NetworkReceive receive;
        //发送出去的响应
        private Send send;
    }
  

4、网络没有建立好会发送消息吗

• 前面分析了消息在发送那个之前会先检查下网络是否建立好，如果没有，会尝试去建立网络，最后如果网络没有建立好，消息是否会发送出去，接下来，我们通过源码来分析下。

• Sedner类当中的run方法核心代码

              /**
               *  步骤四：
               *      检查与要发送数据的主机网络是否建立好
               */
              if (!this.client.ready(node, now)) {
                  //如果网络为建立好，这里返回的就是false， !false就可以进来了
                  //移除这些主机
                  iter.remove();
                  notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
              }
  

• 由于网络没有建立好，它会从迭代器中把这些没有建立好网络连接的主机移除掉，这样就会导致后面的代码无法执行。最后会执行到poll方法

          /**
           *  步骤八：
           *      真正执行网络操作的都是这个NetworkClient组件,作用：发送请求，接受响应
           */
  
          //猜想这里可能会去建立好网络
          this.client.poll(pollTimeout, now);
  

• 结论

  ​ 如果网络没有建立好连接，就移除掉这些主机信息，最后会通过NetworkClient这个网络组建去发送连接请求。然后Sender线程的run不断运行，后面就出现了获取到了网络连接，再次进来到该方法中。执行发送消息的业务逻辑。

5、producer终于与broker建立连接

• 前面介绍了在发送消息之前是需要进行检查网络是否连接好，如果没有连接好，会尝试与服务端建立连接。

• 核心代码回顾

  • NetworkClient.ready方法剖析

        public boolean ready(Node node, long now) {
            //todo: 节点为空就报异常
            if (node.isEmpty())
                throw new IllegalArgumentException("Cannot connect to empty node " + node);
    
            //todo: 判断要发送消息的主机，是否具备发送消息的条件
            if (isReady(node, now))
                return true;
    
            //todo: 第一次进来应该是并没有建立好连接，判断是否可以尝试去建立好网络
            if (connectionStates.canConnect(node.idString(), now))
                // if we are interested in sending to a node and we don't have a connection to it, initiate one
                //todo: 初始化连接
                initiateConnect(node, now);
    
            return false;
        }
    

  • initiateConnect方法剖析，来分析如何与服务端建立连接

        private void initiateConnect(Node node, long now) {
            //todo: 获取节点id
            String nodeConnectionId = node.idString();
            try {
                log.debug("Initiating connection to node {} at {}:{}.", node.id(), node.host(), node.port());
                //todo: 把连接id和当前时间添加到缓存中
                this.connectionStates.connecting(nodeConnectionId, now);
                //todo: 尝试与节点建立socket网络连接
                selector.connect(nodeConnectionId,
                                 new InetSocketAddress(node.host(), node.port()),
                                 this.socketSendBuffer,
                                 this.socketReceiveBuffer);
            } catch (IOException e) {
                /* attempt failed, we'll try again after the backoff */
                //todo: 出现异常把缓存中的连接状态置为DISCONNECTED
                connectionStates.disconnected(nodeConnectionId, now);
                /* maybe the problem is our metadata, update it */
                //todo: 可能由于元数据导致建立网络连接失败，那么更新元数据
                metadataUpdater.requestUpdate();
                log.debug("Error connecting to node {} at {}:{}:", node.id(), node.host(), node.port(), e);
            }
        }
    

• 核心方法selector.connect剖析

  public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
          if (this.channels.containsKey(id))
              throw new IllegalStateException("There is already a connection for id " + id);
  
          /**
           * 如下代码就是一些java NIO编程的基本代码
           */
  
          //获取到SocketChannel
          SocketChannel socketChannel = SocketChannel.open();
          //设置为非阻塞模式
          socketChannel.configureBlocking(false);
          //获取socket
          Socket socket = socketChannel.socket();
          //定期检查一下两边的连接是不是断的
          socket.setKeepAlive(true);
          if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
              //设置socket发送数据的缓存大小
              socket.setSendBufferSize(sendBufferSize);
          if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
              //设置socket接受数据的缓存大小
              socket.setReceiveBufferSize(receiveBufferSize);
          //是否启用nagle算法，默认值false。Negale算法适用于需要发送大量数据的应用场景。这种算法减少传输的次数增加性能
          //它会把网络做的一些小的数据包收集起来，组合成一个大的数据包，再发送出去，
          //但是kafka一定不能把这儿设置为false, 因为有些时候数据包本身就比较小，这个时候就不会发送数据，这是不合理的
          socket.setTcpNoDelay(true);
          boolean connected;
          try {
              //todo: 尝试与服务器连接，由于设置的是非阻塞模式，这种情况下无论操作是否完成都会立刻返回，需要通过其他方式来判断具体操作是否成功。
              connected = socketChannel.connect(address);
          } catch (UnresolvedAddressException e) {
              socketChannel.close();
              throw new IOException("Can't resolve address: " + address, e);
          } catch (IOException e) {
              socketChannel.close();
              throw e;
          }
          //todo: socketChannel向nioSelector注册了一个OP_CONNECT连接事件
          SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
  
          //todo: 通过这个socketChannel构建出KafkaChannel
          KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
  
          //todo: 将key与kafkaChannel关联，方便通过key得到kafkaChannel，以及通过kafkaChannel获取到key
          key.attach(channel);
  
          //todo: 把brokerId与KafkaChannel添加到缓存中
          this.channels.put(id, channel);
  
  
          /**
           * 由于生产者和服务端不在同一台机器上
           * 正常情况下，这儿的网络是不能完成连接的，这里连接不成功，哪里会连接成功呢？ 会在Sender线程中完成
           */
          //todo: 连接成功
          if (connected) {
              // OP_CONNECT won't trigger for immediately connected channels
              log.debug("Immediately connected to node {}", channel.id());
              //添加key到Set<SelectionKey>集合中
              immediatelyConnectedKeys.add(key);
              //取消前面注册 OP_CONNECT 事件
              key.interestOps(0);
          }
      }
  
  

• 疑问

  • 如果尝试与服务端也连接失败了，那么最后到底在哪里与服务端建立的连接？
  • 在Sender线程中完成真正的网络连接

• 核心代码回顾

  • 分析Sender线程中run方法的最后一行处理逻辑

  //完成最后的网络连接       
  this.client.poll(pollTimeout, now);
  

  • poll方法中完成网络连接的核心处理逻辑

    //todo: 2、发送请求、进行复杂的网络操作
              /**
               * 但是我们目前还没有学习到kafka的网络
               * 所以这儿大家就只需要知道这儿会发送网络请求。
               */
              this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
  

  • Selector中的poll方法剖析

   public void poll(long timeout) throws IOException {
          if (timeout < 0)
              throw new IllegalArgumentException("timeout should be >= 0");
  
          clear();
  
          if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty())
              timeout = 0;
  
          /* check ready keys */
          long startSelect = time.nanoseconds();
  
          //todo： 统计Selector上有多少个key注册了
          int readyKeys = select(timeout);
          long endSelect = time.nanoseconds();
          this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());
  
          //todo: 使用场景驱动的方式，刚刚分析了确实有一个key注册在Selector上面了
          if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
              //对Selector上的key进行处理
              pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
              pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
          }
  
          addToCompletedReceives();
  
          long endIo = time.nanoseconds();
          this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
  
          // we use the time at the end of select to ensure that we don't close any connections that
          // have just been processed in pollSelectionKeys
          maybeCloseOldestConnection(endSelect);
      }
  

  • Selector中pollSelectionKeys方法对注册的key进行处理的逻辑剖析

   //todo: 处理Selector上面注册的key
      private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys,
                                     boolean isImmediatelyConnected,
                                     long currentTimeNanos) {
          //todo: 获取到所有的key
          Iterator<SelectionKey> iterator = selectionKeys.iterator();
          //todo: 遍历所有的key
          while (iterator.hasNext()) {
              SelectionKey key = iterator.next();
              iterator.remove();
              //todo: 通过key获取到对应的KafkaChannel
              KafkaChannel channel = channel(key);
  
              // register all per-connection metrics at once
              //todo: 注册连接的监控信息
              sensors.maybeRegisterConnectionMetrics(channel.id());
              if (idleExpiryManager != null)
                  idleExpiryManager.update(channel.id(), currentTimeNanos);
  
              try {
  
                  /* complete any connections that have finished their handshake (either normally or immediately) */
  
                  /**
                   * 代码第一次进来，进入到该分支
                   *    SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
                   */
                  if (isImmediatelyConnected || key.isConnectable()) {
                      //todo: 核心处理逻辑
                      //如果之前都没有完成网络连接，这里会去完成最后的网络连接
                      if (channel.finishConnect()) {
                          //完成网络连接后，会把channel缓存起来
                          this.connected.add(channel.id());
                          this.sensors.connectionCreated.record();
                          SocketChannel socketChannel = (SocketChannel) key.channel();
                          log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",
                                  socketChannel.socket().getReceiveBufferSize(),
                                  socketChannel.socket().getSendBufferSize(),
                                  socketChannel.socket().getSoTimeout(),
                                  channel.id());
                      } else
                          continue;
                  }
  
                  /* if channel is not ready finish prepare */
                  if (channel.isConnected() && !channel.ready())
                      channel.prepare();
  
                  /* if channel is ready read from any connections that have readable data */
                  if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                      NetworkReceive networkReceive;
                      while ((networkReceive = channel.read()) != null)
                          addToStagedReceives(channel, networkReceive);
                  }
  
                  /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
                  if (channel.ready() && key.isWritable()) {
                      Send send = channel.write();
                      if (send != null) {
                          this.completedSends.add(send);
                          this.sensors.recordBytesSent(channel.id(), send.size());
                      }
                  }
  
                  /* cancel any defunct sockets */
                  if (!key.isValid()) {
                      close(channel);
                      this.disconnected.add(channel.id());
                  }
  
              } catch (Exception e) {
                  String desc = channel.socketDescription();
                  if (e instanceof IOException)
                      log.debug("Connection with {} disconnected", desc, e);
                  else
                      log.warn("Unexpected error from {}; closing connection", desc, e);
                  close(channel);
                  this.disconnected.add(channel.id());
              }
          }
      }
  
  

6、producer终于可以发送数据请求了

• 客户端与服务端建立好连接后，客户端底层会通过NIO发送数据请求或者读取响应，这个时候需要向Selector绑定2个事件

  • SelectionKey.OP_READ
    • Socket 读事件，以从远程发送过来了相应数据
  • SelectionKey.OP_WRITE
    • Socket写事件，即向远程发送数据

• 解析下来我们来看看啥时候绑定了OP_READ和OP_WRITE这2个事件

• 核心代码分析

  • Selector类的pollSelectionKeys方法

      //todo: 核心处理逻辑
                        //如果之前都没有完成网络连接，这里会去完成最后的网络连接
                        if (channel.finishConnect()) {
                            //完成网络连接后，会把channel缓存起来
                            this.connected.add(channel.id());
                            this.sensors.connectionCreated.record();
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",
                                    socketChannel.socket().getReceiveBufferSize(),
                                    socketChannel.socket().getSendBufferSize(),
                                    socketChannel.socket().getSoTimeout(),
                                    channel.id());
                        } else
                            continue;
    

  • channel.finishConnect()方法分析

        public boolean finishConnect() throws IOException {
            //todo: 完成最后的网络连接
            boolean connected = socketChannel.finishConnect();
            //todo: 如果连接完成了以后
            if (connected)
                //取消了OP_CONNECT事件
                //增加了OP_READ事件，此时此刻这个key对应的kafkaChannel可以接受到服务端的响应了
                key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
            return connected;
        }
    
    

• 接下来还是回到Sender线程，由于此时连接已经建立好了。由于Sender线程是一个死循环，后面会实现数据的发送

  • run方法核心逻辑

         //todo: 遍历请求
            for (ClientRequest request : requests)
                 //todo:发送请求
                client.send(request, now);
    

  • NetworkClient中的send方法

        public void send(ClientRequest request, long now) {
            String nodeId = request.request().destination();
            if (!canSendRequest(nodeId))
                throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
            //todo: 核心代码
            doSend(request, now);
        }
        private void doSend(ClientRequest request, long now) {
            request.setSendTimeMs(now);
            //todo: 向inFlightRequests组件中添加网络请求
            //表示：发送出去还没有接受到响应的网络请求，默认为5
            //这里我们可以猜想：如果请求发送出去后成功的接受到了响应，然后会把该请求移除掉
            this.inFlightRequests.add(request);
            //todo: 发送请求
            selector.send(request.request());
        }
    

  • Selector中的send方法

        public void send(Send send) {
            //todo: 获取指向对应节点的一个Channel
            KafkaChannel channel = channelOrFail(send.destination());
            try {
                //todo 将send对象放到KafkaChannel.send中去，并没有立刻发送
                channel.setSend(send);
            } catch (CancelledKeyException e) {
                this.failedSends.add(send.destination());
                close(channel);
            }
        }
    

  • KafkaChannel的setSend方法

        public void setSend(Send send) {
            if (this.send != null)
                throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
            //kafkaChannel中缓存发送请求
            this.send = send;
            //todo: 绑定OP_WRITE事件，一旦绑定了该事件，我们就可以向服务端发送请求了
            this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
        }
    
    

    • 此时已经绑定了OP_WRITE事件，接下来我们还是回到Sender中的run方法

    • Sender类

              //todo: 重点就是看这个方法，就是用这个方法拉取kafka集群的元数据
              /**
               *  步骤八：
               *      真正执行网络操作的都是这个NetworkClient组件,作用：发送请求，接受响应
               */
      
              //猜想这里可能会去建立好网络
              this.client.poll(pollTimeout, now);
      

    • NetworkClient中的poll方法

      //发送请求、进行复杂的网络操作
      this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
      

    • Selector中的poll方法

      //核心代码
         //对Selector上的key进行处理
                  pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
      

    • selector中的pollSelectionKeys方法

          //todo: 处理Selector上面注册的key
          private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys,
                                         boolean isImmediatelyConnected,
                                         long currentTimeNanos) {
              //todo: 获取到所有的key
              Iterator<SelectionKey> iterator = selectionKeys.iterator();
              //todo: 遍历所有的key
              while (iterator.hasNext()) {
                  SelectionKey key = iterator.next();
                  iterator.remove();
                  //todo: 通过key获取到对应的KafkaChannel
                  KafkaChannel channel = channel(key);
      
                  // register all per-connection metrics at once
                  //todo: 注册连接的监控信息
                  sensors.maybeRegisterConnectionMetrics(channel.id());
                  if (idleExpiryManager != null)
                      idleExpiryManager.update(channel.id(), currentTimeNanos);
      
                  try {
      
                      /* complete any connections that have finished their handshake (either normally or immediately) */
      
                      /**
                       * 代码第一次进来，进入到该分支
                       *    SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
                       */
                      if (isImmediatelyConnected || key.isConnectable()) {
                          //todo: 核心处理逻辑
                          //如果之前都没有完成网络连接，这里会去完成最后的网络连接
                          if (channel.finishConnect()) {
                              //完成网络连接后，会把channel缓存起来
                              this.connected.add(channel.id());
                              this.sensors.connectionCreated.record();
                              SocketChannel socketChannel = (SocketChannel) key.channel();
                              log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",
                                      socketChannel.socket().getReceiveBufferSize(),
                                      socketChannel.socket().getSendBufferSize(),
                                      socketChannel.socket().getSoTimeout(),
                                      channel.id());
                          } else
                              continue;
                      }
      
                      /* if channel is not ready finish prepare */
                      if (channel.isConnected() && !channel.ready())
                          channel.prepare();
      
                      /* if channel is ready read from any connections that have readable data */
                      //todo: 读取请求
                      if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                          NetworkReceive networkReceive;
                          while ((networkReceive = channel.read()) != null)
                              addToStagedReceives(channel, networkReceive);
                      }
      
                      /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
                      //todo: 处理发送请求的事件
                      if (channel.ready() && key.isWritable()) {
                          //todo: 进行真正的写操作
                          Send send = channel.write();
                          //发送完成
                          if (send != null) {
                              this.completedSends.add(send);
                              this.sensors.recordBytesSent(channel.id(), send.size());
                          }
                      }
      
                      /* cancel any defunct sockets */
                      if (!key.isValid()) {
                          close(channel);
                          this.disconnected.add(channel.id());
                      }
      
                  } catch (Exception e) {
                      String desc = channel.socketDescription();
                      if (e instanceof IOException)
                          log.debug("Connection with {} disconnected", desc, e);
                      else
                          log.warn("Unexpected error from {}; closing connection", desc, e);
                      close(channel);
                      this.disconnected.add(channel.id());
                  }
              }
          }
      

      • KafkaChannel的write方法分析

          public Send write() throws IOException {
              Send result = null;
              //todo: 发送网络请求
              if (send != null && send(send)) {
                  result = send;
                  send = null;
              }
              return result;
          }
      

      • KafkaChannel中的send方法

          private boolean send(Send send) throws IOException {
              //todo: 最终执行发送请求的代码
              send.writeTo(transportLayer);
              //todo: 已经完成请求的发送
              if (send.completed())
                  //移除OP_WRITE事件
                  transportLayer.removeInterestOps(SelectionKey.OP_WRITE);
      
              return send.completed();
          }
      

7、producer是如何处理粘包和拆包问题

• TCP（transport control protocol，传输控制协议）是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有一一成对的socket，因此，发送端为了将多个发往接收端的包，更有效的发到对方，使用了优化方法（Nagle算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样接收端，就难于分辨出来了，必须提供科学的拆包机制。

• 图解TCP的粘包和拆包

  

• 假设客户端分别发送了两个数据包D1和D2给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下四种情况：

  • 1.服务端分两次读取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包
  • 2.服务端一次接受到了两个数据包，D1和D2粘合在一起，称之为TCP粘包
  • 3.服务端分两次读取到了数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，这称之为TCP拆包
  • 4.服务端分两次读取到了数据包，第一次读取到了D1包的部分内容D1_1，第二次读取到了D1包的剩余部分内容D1_2和完整的D2包。
  • 特别要注意的是，如果TCP的接受滑窗非常小，而数据包D1和D2比较大，很有可能会发生第五种情况，即服务端分多次才能将D1和D2包完全接受，期间发生多次拆包。

• 由于producer端在发送消息的时候，一个连接默认是可以最多忍受5个发送出去了没有接受到响应的个数，这就意味着，服务端在发送请求响应的时候，可能一个请求中带有多个响应。这里就涉及到了粘包的问题。

• producer还有可能跟多次读取到服务端的响应的结果，这里就涉及到了拆包的问题

• kafka底层就是通过长度编码方式解决TCP的粘包、拆包问题

• 核心源码分析

  • Selector的pollSelectionKeys方法核心代码

      //todo: 读取请求 ，接受服务端发送回来的响应
     if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
         NetworkReceive networkReceive;
         //todo:不断的接受响应
         while ((networkReceive = channel.read()) != null)
             addToStagedReceives(channel, networkReceive);
     }
    

  • KafkaChannel的read方法

     public NetworkReceive read() throws IOException {
            NetworkReceive result = null;
    
            if (receive == null) {
                //初始化对象NetworkReceive
                receive = new NetworkReceive(maxReceiveSize, id);
            }
    
            //todo:不断接受数据
            receive(receive);
    
            //todo: 是否读完一个完整的消息，这里对拆包问题进行解决
            if (receive.complete()) {
                receive.payload().rewind();
                //赋值receive给result
                result = receive;
                //把receive置为null，方便一下次读取
                receive = null;
            }
            return result;
        }
    

  • receive方法最后会调用NetworkReceive的readFromReadableChannel方法

    //todo: int类型大小的数字(消息体的大小)+ 消息体
        @Deprecated
        public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
            int read = 0;
            //todo: size是一个4字节大小的内存空间
            //todo：如果size还有剩余空间，这个int类型的数字还没有读取完整的结果
            if (size.hasRemaining()) {
                //先读取4字节的数据(代表后面跟着的是消息体的大小)
                int bytesRead = channel.read(size);
                if (bytesRead < 0)
                    throw new EOFException();
                read += bytesRead;
    
                //todo: 一直读取到size没有剩余空间,这里就表示已读取到4字节大小的int类型数字了
                if (!size.hasRemaining()) {
                    size.rewind();
                    //获取int值
                    int receiveSize = size.getInt();
                    if (receiveSize < 0)
                        throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
                    if (maxSize != UNLIMITED && receiveSize > maxSize)
                        throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
    
                    //分配一个内存空间，就是刚刚读取出来4字节的的int类型大小值，然后开辟该内存大小空间用于存储后面的消息体
                    this.buffer = ByteBuffer.allocate(receiveSize);
                }
            }
            if (buffer != null) {
                //读取数据
                int bytesRead = channel.read(buffer);
                if (bytesRead < 0)
                    throw new EOFException();
                read += bytesRead;
            }
    
            return read;
        }
    

  • 解决拆包问题的逻辑NetworkReceive的complete方法

        public boolean complete() {
            //条件：size没有剩余空间和存储消息的内存空间也没有了
            return !size.hasRemaining() && !buffer.hasRemaining();
        }
    
    

    ​

8、如何处理暂存状态的响应

• 前面我们已经可以正常获取到响应的消息了，那生产者是如何处理这些响应信息的？

• 核心代码回顾

  • Sender类中的run方法调用NetworkClient的poll方法

       this.client.poll(pollTimeout, now);
    

  • NetworkClient的poll方法

    //发送网络请求  
    this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    

  • Selector中的poll方法中的核心代码

            if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
                //对Selector上的key进行处理
                pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
                pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
            }
    
    

  • Selector中的pollSelectionKeys方法核心代码

       //todo: 读取请求 ，接受服务端发送回来的响应
                    if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                        NetworkReceive networkReceive;
                        //todo:不断的接受响应
                        //networkReceive就是服务端发送回来的响应
                        while ((networkReceive = channel.read()) != null)
                            //添加响应到队列中
                            addToStagedReceives(channel, networkReceive);
         }
    
    
    

  • Selector中的addToStagedReceives方法核心代码

        private void addToStagedReceives(KafkaChannel channel, NetworkReceive receive) {
            //todo: Map<KafkaChannel, Deque<NetworkReceive>> stagedReceives;
            //已接收完成，但还没有暴露给用户的响应，一个kafkaChannel对应一个队列，一个队列对应一个连接，一个连接会有很多个响应
            //判断map集合中是否包含了该kafkaChannel
    
            //不包含channel这个key就初始化一个队列
            if (!stagedReceives.containsKey(channel))
                //添加队列到map集合中
                stagedReceives.put(channel, new ArrayDeque<NetworkReceive>());
    
            //通过kafkachannel获取该队列
            Deque<NetworkReceive> deque = stagedReceives.get(channel);
            //添加响应到队列中
            deque.add(receive);
        }
    

  • 到此，服务端的请求响应已经被暂存到了stagedReceives这样一个Map<KafkaChannel, Deque>集合中了。

• 响应存储在stagedReceives后，然后进行怎么处理呢，下面我们在来看看核心代码

  • Selector中的poll方法核心代码

       //todo: 对stagedReceives里面的响应进行处理
            addToCompletedReceives();
    

  • Selector中的addToCompletedReceives方法核心代码

      private void addToCompletedReceives() {
            //todo: 如果stagedReceives不为空，有响应数据
            if (!this.stagedReceives.isEmpty()) {
                Iterator<Map.Entry<KafkaChannel, Deque<NetworkReceive>>> iter = this.stagedReceives.entrySet().iterator();
                 //遍历
                while (iter.hasNext()) {
                    Map.Entry<KafkaChannel, Deque<NetworkReceive>> entry = iter.next();
                    //获取kafkachannel
                    KafkaChannel channel = entry.getKey();
    
                    if (!channel.isMute()) {
                        //todo：获取响应队列
                        Deque<NetworkReceive> deque = entry.getValue();
                        //todo: 从队列中取出响应
                        NetworkReceive networkReceive = deque.poll();
                        //todo: 添加响应到已经处理完成的队列中
                        this.completedReceives.add(networkReceive);
                        this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
                        if (deque.isEmpty())
                            iter.remove();
                    }
                }
            }
        }
    

• 目前响应仅仅只是保存起来了，还没有对响应数据进行处理，接下来我们就来看看对响应数据的处理

  • NetworkClient中的poll方法分析

     public List<ClientResponse> poll(long timeout, long now) {
    
            //todo: 1、封装了一个要拉取元数据请求
            long metadataTimeout = metadataUpdater.maybeUpdate(now);
            try {
                //todo: 2、发送请求、进行复杂的网络操作
                /**
                 * 但是我们目前还没有学习到kafka的网络
                 * 所以这儿大家就只需要知道这儿会发送网络请求。
                 */
                this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
            } catch (IOException e) {
                log.error("Unexpected error during I/O", e);
            }
    
            // process completed actions
            long updatedNow = this.time.milliseconds();
            List<ClientResponse> responses = new ArrayList<>();
              //处理完成的发送响应
            handleCompletedSends(responses, updatedNow);
            /**
             * 步骤三：处理响应，响应里面就会有我们需要的元数据。
             *
             * 这个地方是我们在看生产者是如何获取元数据的时候看的。
             * 其实Kafak获取元数据的流程跟我们发送消息的流程是一模一样。
             * 获取元数据 ---> 判断网络连接是否建立好 ---> 建立网络连接 ---> 发送请求（获取元数据的请求）---> 服务端发送回来响应（带了集群的元数据信息）
             */
    
            //todo:处理完成的接收响应--- 核心代码
            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);
                    }
                }
            }
    
            return responses;
        }
    
    

  • NetworkClient中的handleCompletedReceives方法分析

        //todo: 对响应进行一定的处理
        private void handleCompletedReceives(List<ClientResponse> responses, long now) {
            //todo: 遍历每一个已经完成响应
            for (NetworkReceive receive : this.selector.completedReceives()) {
                //返回响应
                String source = receive.source();
                //从inFlightRequests中取出对应的ClientRequest
                //kafka有这样一个发送请求的机制：默认每个连接最多容忍5个发送出去还没有接受到响应的请求
    
                //todo:  从InFlightRequests中移除掉接受到响应的请求
                ClientRequest req = inFlightRequests.completeNext(source);
                //解析响应,数据封装成Struct对象中
                Struct body = parseResponse(receive.payload(), req.request().header());
    
                //todo: 如果是关于元数据的响应
                if (!metadataUpdater.maybeHandleCompletedReceive(req, now, body))
                    //解析完了以后封装成一个一个的ClientResponse对象
                    //body：表示存储响应的内容
                    //req： 表示发送的请求
                    responses.add(new ClientResponse(req, now, false, body));
            }
        }
    
    

  • kafka响应消息的流转图

    

​

9、如何处理响应消息

• 客户端接受到服务端的请求响应，最后转换成了ClientResponse，接下来分析下生产者如何处理响应ClientResponse

• 核心代码回顾

  • Sender类中的run方法调用NetworkClient的poll方法

    public List<ClientResponse> poll(long timeout, long now) {
    
            //todo: 1、封装了一个要拉取元数据请求
            long metadataTimeout = metadataUpdater.maybeUpdate(now);
            try {
                //todo: 2、发送请求、进行复杂的网络操作
                /**
                 * 但是我们目前还没有学习到kafka的网络
                 * 所以这儿大家就只需要知道这儿会发送网络请求。
                 */
                this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
            } catch (IOException e) {
                log.error("Unexpected error during I/O", e);
            }
    
            // process completed actions
            long updatedNow = this.time.milliseconds();
            List<ClientResponse> responses = new ArrayList<>();
            //处理完成的发送响应
            handleCompletedSends(responses, updatedNow);
            /**
             * 步骤三：处理响应，响应里面就会有我们需要的元数据。
             *
             * 这个地方是我们在看生产者是如何获取元数据的时候看的。
             * 其实Kafak获取元数据的流程跟我们发送消息的流程是一模一样。
             * 获取元数据 ---> 判断网络连接是否建立好 ---> 建立网络连接 ---> 发送请求（获取元数据的请求）---> 服务端发送回来响应（带了集群的元数据信息）
             */
    
            //todo: 处理完成的接收响应  ---核心代码
            handleCompletedReceives(responses, updatedNow);
            //处理关闭的响应
            handleDisconnections(responses, updatedNow);
            //处理连接状态变化
            handleConnections();
            //处理超时的请求
            handleTimedOutRequests(responses, updatedNow);
    
            //todo: 对ClientResponse进行处理--------->本次分析的重点
            // invoke callbacks
            for (ClientResponse response : responses) {
                if (response.request().hasCallback()) {
                    try {
                        //todo: 调用了响应中的(之前发送出去的)请求的回调函数
                        //由于我们之前在封装网络请求的过程绑定了回调函数，这里是调用了request的回调函数，
                        response.request().callback().onComplete(response);
                    } catch (Exception e) {
                        log.error("Uncaught error in request completion:", e);
                    }
                }
            }
    
            return responses;
        }
    
    

    

  • Sender线程中的run方法核心代码分析

      //todo:如果网络没有建立好，这块代码也是不会执行的
      List<ClientRequest> requests = createProduceRequests(batches, now);
    
        /**
         * Transfer the record batches into a list of produce requests on a per-node basis
         */
        private List<ClientRequest> createProduceRequests(Map<Integer, List<RecordBatch>> collated, long now) {
            List<ClientRequest> requests = new ArrayList<ClientRequest>(collated.size());
            for (Map.Entry<Integer, List<RecordBatch>> entry : collated.entrySet())
                requests.add(produceRequest(now, entry.getKey(), acks, requestTimeout, entry.getValue()));
            return requests;
        }
    

  • Sender线程中的produceRequest方法代码分析

    /**
         * Create a produce request from the given record batches
         * todo: 从批次中封装请求
         */
        private ClientRequest produceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches) {
            Map<TopicPartition, ByteBuffer> produceRecordsByPartition = new HashMap<TopicPartition, ByteBuffer>(batches.size());
            final Map<TopicPartition, RecordBatch> recordsByPartition = new HashMap<TopicPartition, RecordBatch>(batches.size());
            for (RecordBatch batch : batches) {
                TopicPartition tp = batch.topicPartition;
                produceRecordsByPartition.put(tp, batch.records.buffer());
                recordsByPartition.put(tp, batch);
            }
            //todo:基于每一个批次封装request请求
            ProduceRequest request = new ProduceRequest(acks, timeout, produceRecordsByPartition);
            //封装发送数据的请求
            RequestSend send = new RequestSend(Integer.toString(destination),
                                               this.client.nextRequestHeader(ApiKeys.PRODUCE),
                                               request.toStruct());
    
            //todo: 定义回调函数
            RequestCompletionHandler callback = new RequestCompletionHandler() {
                public void onComplete(ClientResponse response) {
                    //todo: 回调函数处理的业务逻辑
                    handleProduceResponse(response, recordsByPartition, time.milliseconds());
                }
            };
    
            //todo: 请求绑定回调函数
            return new ClientRequest(now, acks != 0, send, callback);
        }
    

  • Sender线程中的handleProduceResponse方法代码分析

     /**
         * Handle a produce response
         *
         * todo: 执行回调函数的处理逻辑
         */
        private void handleProduceResponse(ClientResponse response, Map<TopicPartition, RecordBatch> batches, long now) {
            int correlationId = response.request().request().header().correlationId();
            //todo: 发送请求时，broker可能失去连接，这是一个小概率事件
            if (response.wasDisconnected()) {
                log.trace("Cancelled request {} due to node {} being disconnected", response, response.request()
                                                                                                      .request()
                                                                                                      .destination());
                for (RecordBatch batch : batches.values())
                    completeBatch(batch, Errors.NETWORK_EXCEPTION, -1L, Record.NO_TIMESTAMP, correlationId, now);
            } else {
                log.trace("Received produce response from node {} with correlation id {}",
                          response.request().request().destination(),
                          correlationId);
                // if we have a response, parse it
    
                //todo: 正常情况下，进入到该分支
                if (response.hasResponse()) {
                    ProduceResponse produceResponse = new ProduceResponse(response.responseBody());
                    //todo: 遍历每个分区的响应
                    for (Map.Entry<TopicPartition, ProduceResponse.PartitionResponse> entry : produceResponse.responses().entrySet()) {
                        //todo: 获取每一个分区对象TopicPartition
                        TopicPartition tp = entry.getKey();
                        //todo: 获取每一个分区的响应
                        ProduceResponse.PartitionResponse partResp = entry.getValue();
                        //todo: 服务器处理失败了，返回异常信息
                        Errors error = Errors.forCode(partResp.errorCode);
    
                         //todo: 获取分区的数据
                        RecordBatch batch = batches.get(tp);
    
                         //todo: 对响应进行处理
                        completeBatch(batch, error, partResp.baseOffset, partResp.timestamp, correlationId, now);
                    }
                    this.sensors.recordLatency(response.request().request().destination(), response.requestLatencyMs());
                    this.sensors.recordThrottleTime(response.request().request().destination(),
                                                    produceResponse.getThrottleTime());
                } else {
                    // this is the acks = 0 case, just complete all requests
                    /**
                     * todo: acks=0  不需要返回响应
                     *
                     * 在实际的生产环境中一般不会设置acks=0
                     */
                    for (RecordBatch batch : batches.values())
                        completeBatch(batch, Errors.NONE, -1L, Record.NO_TIMESTAMP, correlationId, now);
                }
            }
        }
    

  • Sender线程中的completeBatch方法代码分析

     private void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now) {
            /**
             * todo: 如果响应中带有异常，并且该请求时可以重试的
             */
            if (error != Errors.NONE && canRetry(batch, error)) {
                // retry
                log.warn("Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}",
                         correlationId,
                         batch.topicPartition,
                         this.retries - batch.attempts - 1,
                         error);
                this.accumulator.reenqueue(batch, now);
                this.sensors.recordRetries(batch.topicPartition.topic(), batch.recordCount);
            } else {
                /**
                 * todo: 如果带有异常 或者 不允许重试
                 */
                RuntimeException exception;
                //todo: 带有权限的异常
                if (error == Errors.TOPIC_AUTHORIZATION_FAILED)
                    //todo: 自定义权限异常
                    exception = new TopicAuthorizationException(batch.topicPartition.topic());
                else
                    //todo: 非权限的异常直接获取异常
                    exception = error.exception();
                // tell the user the result of their request
                /**
                 *  todo: 核心代码,对异常进行处理
                 */
                batch.done(baseOffset, timestamp, exception);
                this.accumulator.deallocate(batch);
                if (error != Errors.NONE)
                    this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);
            }
            if (error.exception() instanceof InvalidMetadataException) {
                if (error.exception() instanceof UnknownTopicOrPartitionException)
                    log.warn("Received unknown topic or partition error in produce request on partition {}. The " +
                            "topic/partition may not exist or the user may not have Describe access to it", batch.topicPartition);
                metadata.requestUpdate();
            }
    
            // Unmute the completed partition.
            if (guaranteeMessageOrder)
                this.accumulator.unmutePartition(batch.topicPartition);
        }
    

  • RecordBatch中的done方法代码分析

    public void done(long baseOffset, long timestamp, RuntimeException exception) {
            log.trace("Produced messages to topic-partition {} with base offset offset {} and error: {}.",
                      topicPartition,
                      baseOffset,
                      exception);
            /**
             * todo: 一条消息就代表一个Thunk
             */
            // execute callbacks
            for (int i = 0; i < this.thunks.size(); i++) {
                try {
                    Thunk thunk = this.thunks.get(i);
                    //todo: 如果没有异常
                    if (exception == null) {
                        // If the timestamp returned by server is NoTimestamp, that means CreateTime is used. Otherwise LogAppendTime is used.
                        RecordMetadata metadata = new RecordMetadata(this.topicPartition,  baseOffset, thunk.future.relativeOffset(),
                                                                     timestamp == Record.NO_TIMESTAMP ? thunk.future.timestamp() : timestamp,
                                                                     thunk.future.checksum(),
                                                                     thunk.future.serializedKeySize(),
                                                                     thunk.future.serializedValueSize());
    
                        //todo: 把无异常信息传递给我们在通过生产者发送消息的过程中绑定的回调函数
                        thunk.callback.onCompletion(metadata, null);
                    } else {
                       //todo: 如果有异常 ，调用绑定的回调函数，把异常信息传递给回调函数，
                       //todo: 这样生产者代码中就可以获取到异常信息了，然后结合公司的业务规则对异常信息进行处理就可以了
                        thunk.callback.onCompletion(null, exception);
                    }
                } catch (Exception e) {
                    log.error("Error executing user-provided callback on message for topic-partition {}:", topicPartition, e);
                }
            }
            this.produceFuture.done(topicPartition, baseOffset, exception);
        }
    

  • 回到我们写的生产者发送代码逻辑

    

    

10、 消息发送完了以后内存如何处理

• 上面我们已经分析到了通过回调函数对响应进行处理，基本上到这一块发送消息的流程已经结束了，那么消息发送出去后，存储RecordBath数据的内存如何处理，下面我们来学习下

• Sender线程中的completeBatch方法核心代码分析

    /**
      *  todo: 核心代码,对异常进行处理,里面调用了用户传进来的回调函数
      */
   batch.done(baseOffset, timestamp, exception);
  
     //todo: 回收内存资源
    this.accumulator.deallocate(batch);
  

• RecordAccumulator的deallocate方法剖析

      /**
       * Deallocate the record batch
       */
      public void deallocate(RecordBatch batch) {
          //todo: 从IncompleteRecordBatches中移除成功处理的批次
          incomplete.remove(batch);
  
          //todo: 释放内存
          free.deallocate(batch.records.buffer(), batch.records.initialCapacity());
      }
  

• BufferPool的deallocate方法剖析

  • 之前分析过的逻辑

  public void deallocate(ByteBuffer buffer, int size) {
          //加锁
          lock.lock();
          try {
              //todo:需要申请的内存等于一个批次的内存大小，就把内存归还给内存池
              //该判断可以有效的提高内存的利用率，等于批次大小的内存直接归还内存池利用率才是最高的
              if (size == this.poolableSize && size == buffer.capacity()) {
                  //todo: 清空数据
                  buffer.clear();
                  //todo:内存放入到内存池中
                  this.free.add(buffer);
              } else {
              //todo:否则不相等，就直接归还给可用的内存中
                  this.availableMemory += size;
              }
              Condition moreMem = this.waiters.peekFirst();
              if (moreMem != null)
                  //todo: 内存释放后，需要唤醒处于wait等待内存的线程----> !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
                  //唤醒正在等待分配内存的线程
                  moreMem.signal();
          } finally {
              lock.unlock();
          }
      }
  

11、 消息有异常是如何处理的

• 上面我们已经分析到了消息被处理成功后，会进行内存回收。如果发送消息的过程中服务端有异常，在响应中会封装异常信息，返回给客户端，下面我们再来具体学习下如何处理这些异常信息的。

• Sender线程中的completeBatch方法核心代码分析

   private void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now) {
          /**
           * todo: 如果响应中带有异常，并且该请求时可以重试的
           */
          if (error != Errors.NONE && canRetry(batch, error)) {
              // retry
              log.warn("Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}",
                       correlationId,
                       batch.topicPartition,
                       this.retries - batch.attempts - 1,
                       error);
  
              //todo: 把发送失败的批次重新加入到队列中
              this.accumulator.reenqueue(batch, now);
              this.sensors.recordRetries(batch.topicPartition.topic(), batch.recordCount);
          } else {
              /**
               * todo: 如果带有异常 或者 不允许重试
               */
              RuntimeException exception;
              //todo: 带有权限的异常
              if (error == Errors.TOPIC_AUTHORIZATION_FAILED)
                  //todo: 自定义权限异常
                  exception = new TopicAuthorizationException(batch.topicPartition.topic());
              else
                  //todo: 非权限的异常直接获取异常
                  exception = error.exception();
              // tell the user the result of their request
              /**
               *  todo: 核心代码,对异常进行处理,里面调用了用户传进来的回调函数
               */
              batch.done(baseOffset, timestamp, exception);
  
              //todo: 回收内存资源
              this.accumulator.deallocate(batch);
              if (error != Errors.NONE)
                  this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);
          }
          if (error.exception() instanceof InvalidMetadataException) {
              if (error.exception() instanceof UnknownTopicOrPartitionException)
                  log.warn("Received unknown topic or partition error in produce request on partition {}. The " +
                          "topic/partition may not exist or the user may not have Describe access to it", batch.topicPartition);
              metadata.requestUpdate();
          }
  
          // Unmute the completed partition.
          if (guaranteeMessageOrder)
              this.accumulator.unmutePartition(batch.topicPartition);
      }
  

• RecordAccumulator的reenqueue方法剖析

      /**
       * Re-enqueue the given record batch in the accumulator to retry
       */
      public void reenqueue(RecordBatch batch, long now) {
          //todo: 重试次数+1
          batch.attempts++;
          //todo: 上一次重试的时间等于当前时间
          batch.lastAttemptMs = now;
          //todo: 上一次添加到队列中的时间
          batch.lastAppendTime = now;
          batch.setRetry();
  
          //todo: 如果之前该批次对应的分区有队列就使用之前的队列，没有就重新创建一个
          Deque<RecordBatch> deque = getOrCreateDeque(batch.topicPartition);
          synchronized (deque) {
              //todo: 添加批次到队列的头部
              deque.addFirst(batch);
          }
      }
  

• 生产者代码中的发送数据的回调函数

  

  

• 小结

  对异常处理分类：

  （1）有异常并且还有可重试的次数

  ​ 策略：进行重试

  （2）无异常或者带有异常或者没有重试次数

  ​ a. 如果是带有权限异常，就自定义封装权限异常类

  ​ b. 如果没有带有权限异常（可能是其他异常或者无异常）

  ​ 调用生产者发送数据指定的回调函数封装可能包含的异常信息，通过回调函数，获取到后进行处理判断

12、 如何处理超时的批次

• Sender线程的run方法运行后，其中会对队列中发送的批次进行检测，看看该批次是否超时，如果出现了超时，会采取一定的策略处理，下面我们基于源码来分析下

• Sender线程的run方法批次超时核心代码

   /**
           * TODO: 步骤六：
           * todo: 放弃超时的batches
           * 超时批次的处理逻辑
           */
          List<RecordBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);
  

• RecordAccumulator的abortExpiredBatches方法分析

      public List<RecordBatch> abortExpiredBatches(int requestTimeout, long now) {
          //todo: 定义一个数组用来存储超时的批次
          List<RecordBatch> expiredBatches = new ArrayList<>();
          int count = 0;
          //todo: 遍历每个队列
          for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {
              //todo: 获取每个分区的队列
              Deque<RecordBatch> dq = entry.getValue();
              //todo：获取队列的分区信息
              TopicPartition tp = entry.getKey();
              // We only check if the batch should be expired if the partition does not have a batch in flight.
              // This is to prevent later batches from being expired while an earlier batch is still in progress.
              // Note that `muted` is only ever populated if `max.in.flight.request.per.connection=1` so this protection
              // is only active in this case. Otherwise the expiration order is not guaranteed.
              //todo: 没有正在处理的分区批次
              if (!muted.contains(tp)) {
                  synchronized (dq) {
                      // iterate over the batches and expire them if they have been in the accumulator for more than requestTimeOut
                      //获取指定分区队列最后一个批次
                      RecordBatch lastBatch = dq.peekLast();
                      Iterator<RecordBatch> batchIterator = dq.iterator();
                      //迭代队列中的每一个批次
                      while (batchIterator.hasNext()) {
                          //获取批次
                          RecordBatch batch = batchIterator.next();
                          boolean isFull = batch != lastBatch || batch.records.isFull();
                          //todo: 判断批次是否超时
                          // check if the batch is expired
                          if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) {
                              //todo: 保存超时的批次
                              expiredBatches.add(batch);
                              //todo: 超时批次计数累加1
                              count++;
                              //todo: 从数据结构中移除超时的批次
                              batchIterator.remove();
                              //todo: 释放内存资源
                              deallocate(batch);
                          } else {
                              // Stop at the first batch that has not expired.
                              break;
                          }
                      }
                  }
              }
          }
          if (!expiredBatches.isEmpty())
              log.trace("Expired {} batches in accumulator", count);
  
          return expiredBatches;
      }
  

• RecordBatch的maybeExpire批次超时检测方法

      public boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) {
          boolean expire = false;
          String errorMessage = null;
          /**
           * requestTimeoutMs:请求发送超时时间，默认是30s
           * todo: 没设置重试，并且发送批次（batch.size）满了，并且配置请求超时时间（request.timeout.ms）小于【当前时间减去最后追加批次的时间】
           */
          if (!this.inRetry() && isFull && requestTimeoutMs < (now - this.lastAppendTime)) {
              //超时的标识
              expire = true;
              //记录异常信息
              errorMessage = (now - this.lastAppendTime) + " ms has passed since last append";
          //todo: 没设置重试，并且配置请求超时时间（request.timeout.ms）小于【创建批次时间减去配置的等待发送的时间（linger.ms）】
          } else if (!this.inRetry() && requestTimeoutMs < (now - (this.createdMs + lingerMs))) {
              expire = true;
              errorMessage = (now - (this.createdMs + lingerMs)) + " ms has passed since batch creation plus linger time";
          //todo: 设置重试，并且配置请求超时时间（request.timeout.ms）小于【当前时间-最后重试时间-重试需要等待的时间（retry.backoff.ms）】
          } else if (this.inRetry() && requestTimeoutMs < (now - (this.lastAttemptMs + retryBackoffMs))) {
              expire = true;
              errorMessage = (now - (this.lastAttemptMs + retryBackoffMs)) + " ms has passed since last attempt plus backoff time";
          }
  
  
          //todo: 如果超时
          if (expire) {
              this.records.close();
              //todo: 调用done方法，方法中封装TimeoutException超时异常
              this.done(-1L, Record.NO_TIMESTAMP,
                        new TimeoutException("Expiring " + recordCount + " record(s) for " + topicPartition + " due to " + errorMessage));
          }
  
          return expire;
      }
  

13、 如何处理长时间没有接受到响应的消息

• 最后我们来分析下生产者如果长时间没有接受到服务端的响应，对于这种请求来说，是如何处理的？

• Sender线程的run方法运行后，其中会对队列中发送的批次进行检测，看看该批次是否超时，如果出现了超时，会采取一定的策略处理，下面我们基于源码来分析下

• NetworkClient的poll方法核心代码

    //处理超时的请求
   handleTimedOutRequests(responses, updatedNow);
  

• NetworkClient的handleTimedOutRequests方法核心代码

   private void handleTimedOutRequests(List<ClientResponse> responses, long now) {
          //todo: 获取超时的主机列表
          List<String> nodeIds = this.inFlightRequests.getNodesWithTimedOutRequests(now, this.requestTimeoutMs);
          //todo：遍历
          for (String nodeId : nodeIds) {
              // close connection to the node
              //todo: 关闭请求超时的主机连接
              this.selector.close(nodeId);
              log.debug("Disconnecting from node {} due to request timeout.", nodeId);
              //todo: 修改连接状态
              processDisconnection(responses, nodeId, now);
          }
  
          // we disconnected, so we should probably refresh our metadata
          if (nodeIds.size() > 0)
              metadataUpdater.requestUpdate();
      }
  

• InFlightRequests的getNodesWithTimedOutRequests获取请求超时的主机列表

      public List<String> getNodesWithTimedOutRequests(long now, int requestTimeout) {
          List<String> nodeIds = new LinkedList<>();
          //todo：遍历
          for (Map.Entry<String, Deque<ClientRequest>> requestEntry : requests.entrySet()) {
              //todo:获取主机名
              String nodeId = requestEntry.getKey();
              //todo: 获取对应的请求队列
              Deque<ClientRequest> deque = requestEntry.getValue();
  
              //todo: 队列中有请求
              if (!deque.isEmpty()) {
                  //todo: 从队列中取出最早的一个请求
                  ClientRequest request = deque.peekLast();
                  //todo: 该请求发送了多久
                  long timeSinceSend = now - request.sendTimeMs();
                  //todo: 如果超时
                  if (timeSinceSend > requestTimeout)
                      //todo: 添加超时的主机到LinkedList集合中
                      nodeIds.add(nodeId);
              }
          }
  
          //todo:返回超时的主机列表
          return nodeIds;
      }
  

• NetworkClient的processDisconnection方法修改连接状态

      private void processDisconnection(List<ClientResponse> responses, String nodeId, long now) {
          //todo； 修改连接状态
          connectionStates.disconnected(nodeId, now);
          //todo：从inFlightRequests中清空超时的请求
          for (ClientRequest request : this.inFlightRequests.clearAll(nodeId)) {
              log.trace("Cancelled request {} due to node {} being disconnected", request, nodeId);
              if (!metadataUpdater.maybeHandleDisconnection(request))
                  //todo: 对这些请求进行处理，它是自己封装了一个响应，响应中没有响应消息为null, 失去连接的标识为true
                  responses.add(new ClientResponse(request, now, true, null));
          }
      }
  

14、生产者源码精华总结

• 总结下KafkaProducer生产者源码哪些地方值得我们学习
  • 1、kafka网络部分的设计绝对是一个亮点，kafka自己基于NIO封装了一套自己的网络通信框架，支持一个客户端与多个broker建立连接
  • 2、处理拆包和粘包的的思路和代码，绝对是教科书级别的，可以把代码复制粘贴下来直接用到自己的项目中去。
  • 3、RecordAccumulator封装消息的batchs，使用的自己封装的数据结构CopyOnWriteMap，采用读写分离的思想，用来面对高并发的场景（读多，写少）提升整个流程的性能。
  • 4、同时封装消息的时候设计的内存缓冲池，极大的减少了GC的次数。
  • 5、RecordAccumulator封装批次代码中采用的是分段加锁的思想，极大的提高了性能。