2.1.2 客户端消息发送线程

╰半夏微凉° 2023-01-12 15:53 19阅读 0赞

--------------------

#### 2.1.2 客户端消息发送线程 ####

> 追加消息到记录收集器时按照分区进行分组，并放到 batches集合中，每个分区的队列都保存了即将发送到这个分区对应节点上的批记录，客户端的发送线程可 以只使用一个 Sender线程迭代batches的每个分区，获取分区对应的主剧本节点，取出分区对应的队列中的批记录就可以发送消息了 。

消息发送线程有两种消息发送方式：按照分区直接发送 、 按照分区的目标节点发迭。 假设有两台服务器， 主题有6个分区，那么每台服务器就有3个分区。 如图2-7 (左)所示，消息发送线程迭代batches的每个分区，直接往分区的主副本节点发送消息，总共会有6个请求。如图2-7 (右)所示，我们先按照分区的主副本节点进行分组，把属于同一个节点的所有分区放在一起，总共只有两个请求。第二种做法可以大大减少网络的开销。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

1.  从记录收集器获取数据  
    生产者发送的消息在客户端首先被保存到记录收集器中，发送线程需要发送消息时，从中获取就可以了。不过记录收集器并不仅仅将消息暂存起来， 而且为了使发送线程能够更好地工作，追加到记录收集器的消息将按照分区放好。在发送线程需要数据时，记录收集器能够按照节点将消息重新分组再交给发送线程。发送线程从记录收集器中得到每个节点上需要发送的批记录列表 ， 为每个节点都创建一个客户端请求(ClientRequest )。相关代码如下 ：

private long sendProducerData(long now) { 
            Cluster cluster = metadata.fetch();
            // get the list of partitions with data ready to send
            RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
    
            // if there are any partitions whose leaders are not known yet, force metadata update
            if (!result.unknownLeaderTopics.isEmpty()) { 
                // The set of topics with unknown leader contains topics with leader election pending as well as
                // topics which may have expired. Add the topic again to metadata to ensure it is included
                // and request metadata update, since there are messages to send to the topic.
                for (String topic : result.unknownLeaderTopics)
                    this.metadata.add(topic);
                this.metadata.requestUpdate();
            }
    
            // remove any nodes we aren't ready to send to
            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));
                }
            }
    
            // create produce requests
            Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes,
                    this.maxRequestSize, now);
            if (guaranteeMessageOrder) { 
                // Mute all the partitions drained
                for (List<ProducerBatch> batchList : batches.values()) { 
                    for (ProducerBatch batch : batchList)
                        this.accumulator.mutePartition(batch.topicPartition);
                }
            }
    
            List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(this.requestTimeout, now);
            boolean needsTransactionStateReset = false;
            // Reset the producer id if an expired batch has previously been sent to the broker. Also update the metrics
            // for expired batches. see the documentation of @TransactionState.resetProducerId to understand why
            // we need to reset the producer id here.
            if (!expiredBatches.isEmpty())
                log.trace("Expired {} batches in accumulator", expiredBatches.size());
            for (ProducerBatch expiredBatch : expiredBatches) { 
                failBatch(expiredBatch, -1, NO_TIMESTAMP, expiredBatch.timeoutException());
                if (transactionManager != null && expiredBatch.inRetry()) { 
                    needsTransactionStateReset = true;
                }
                this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);
            }
    
            if (needsTransactionStateReset) { 
                transactionManager.resetProducerId();
                return 0;
            }
    
            sensors.updateProduceRequestMetrics(batches);
    
            // If we have any nodes that are ready to send + have sendable data, poll with 0 timeout so this can immediately
            // loop and try sending more data. Otherwise, the timeout is determined by nodes that have partitions with data
            // that isn't yet sendable (e.g. lingering, backing off). Note that this specifically does not include nodes
            // with sendable data that aren't ready to send since they would cause busy looping.
            long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
            if (!result.readyNodes.isEmpty()) { 
                log.trace("Nodes with data ready to send: {}", result.readyNodes);
                // if some partitions are already ready to be sent, the select time would be 0;
                // otherwise if some partition already has some data accumulated but not ready yet,
                // the select time will be the time difference between now and its linger expiry time;
                // otherwise the select time will be the time difference between now and the metadata expiry time;
                pollTimeout = 0;
            }
            sendProduceRequests(batches, now);
            return pollTimeout;
        }

追加消息到记录收集器的数据结构是batches : TopicPartition → Deque<RecordBatch＞，读取记录收集器的数据结构是 batches:Nodeld → Li.st<RecordBatch＞。 为了区分这两个 batches，把后者叫作  
batches， 从batches转变为batches的步骤如下 。

1.  迭代 batches 的每个分区，获取TopicPartition应的主副本节点 ： NodeId 。
2.  获取分区的批记录队列中的第一个批记录 ： TopicPartition → RecordBatch 。
3.  将相同主副本节点的所有分区放在一起 ： NodeId → List。
4.  将相同主副本节点的所有批记录放在一起： NodeId → List< RecordBatch> 。

如图2 -8所示，步骤（ 1 ）产生batches ，步骤（4）产生batches ’ 。 发送线程从记录收集器获取数据，然  
后创建客户端请求并发送给服务端，具体步骤如下 。

1.  消息被记录收集器收集，并按照分区追加到队列的最后一个批记录中 。
2.  发送钱程通过ready()从记录收集器中找出已经准备好的服务端节点 。
3.  节点已经准备好 ， 如果客户端还没有和 它们建立连接，通过 connect()建立到服务端的连接 。
4.  发送钱程通过drain()从记录收集器获取按照节点整理好的每个分区的批记录 。
5.  发送线程得到每个节点的批记录后 ， 为每个节点创建客户端请求，并将请求发送到服务端 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
发送线程不仅要从记录收集器读取数据 ， 而且还要将读取到的数据用来创建客户端请求。

1.  创建生产者客户端请求  
    从记录收集器获取出来的 batches 已经按照节点分组了 ， 发送线程的produceRequest()方法会为每个节点都创建一个客户端请求 。 ProduceRequest ()方法的 batches参数是指定目标节点的批记录列表。由于一个目标节点就有多个分区，每个批记录都对应了一个分区 。 创建客户端请求时， 批记录列表需要转成分区到字节缓冲区的字典结构。 相关代码如下：

private void sendProduceRequest(long now, int destination, short acks, int timeout, List<ProducerBatch> batches) { 
            if (batches.isEmpty())
                return;
    
            Map<TopicPartition, MemoryRecords> produceRecordsByPartition = new HashMap<>(batches.size());
            final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
    
            // find the minimum magic version used when creating the record sets
            byte minUsedMagic = apiVersions.maxUsableProduceMagic();
            for (ProducerBatch batch : batches) { 
                if (batch.magic() < minUsedMagic)
                    minUsedMagic = batch.magic();
            }
    
            for (ProducerBatch batch : batches) { 
                TopicPartition tp = batch.topicPartition;
                MemoryRecords records = batch.records();
    
                // down convert if necessary to the minimum magic used. In general, there can be a delay between the time
                // that the producer starts building the batch and the time that we send the request, and we may have
                // chosen the message format based on out-dated metadata. In the worst case, we optimistically chose to use
                // the new message format, but found that the broker didn't support it, so we need to down-convert on the
                // client before sending. This is intended to handle edge cases around cluster upgrades where brokers may
                // not all support the same message format version. For example, if a partition migrates from a broker
                // which is supporting the new magic version to one which doesn't, then we will need to convert.
                if (!records.hasMatchingMagic(minUsedMagic))
                    records = batch.records().downConvert(minUsedMagic, 0);
                produceRecordsByPartition.put(tp, records);
                recordsByPartition.put(tp, batch);
            }
    
            String transactionalId = null;
            if (transactionManager != null && transactionManager.isTransactional()) { 
                transactionalId = transactionManager.transactionalId();
            }
            ProduceRequest.Builder requestBuilder = new ProduceRequest.Builder(minUsedMagic, acks, timeout,
                    produceRecordsByPartition, transactionalId);
            RequestCompletionHandler callback = new RequestCompletionHandler() { 
                public void onComplete(ClientResponse response) { 
                    handleProduceResponse(response, recordsByPartition, time.milliseconds());
                }
            };
    
            String nodeId = Integer.toString(destination);
            ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0, callback);
            client.send(clientRequest, now);
            log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
        }

这里需要注意的是，`发送线程并不负责真正发送客户端请求， 它会从记录收集器中取出要发送的消息，创建好客户端请求，然后把请求交给客户端网络对象(NetworkClient)去发送`。因为没有在发送线程中发送请求，所以创建客户端请求时需要保留目标节点，这样客户端网络对象获取出客户端请求时，才能知道要发送给哪个目标节点 。

--------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/ae1797c22ba344c39198ff92e39c5c79.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/cb9e2426b3be4e69b212a14def262880.png