Kafka源码分析-07 | 真正发送消息的组件——Sender分析

假设有两台服务端机器，一个主题有6个分区，每台机器有3个分区。如果按照分区发送就会有6个请求，而按节点发送总共就有两个请求。按节点发送可以大大减少网络的开销。

6. 收集和处理过期的批次。在有些情况下，消息很久都没发送出去，我们不可能让消息批次一直等待，要有对应的超时批次的处理逻辑。先看下详细代码：

private List

getExpiredInflightBatches(long now) { //1.创建过期批次集合 List

expiredBatches = new ArrayList<>(); //2.遍历inFlightBatches for (Iterator>> batchIt = inFlightBatches.entrySet().iterator(); batchIt.hasNext();) { Map.Entry> entry = batchIt.next(); List

partitionInFlightBatches = entry.getValue(); if (partitionInFlightBatches != null) { Iterator

iter = partitionInFlightBatches.iterator(); //3.遍历某个分区的生产者批量发送列表 while (iter.hasNext()) { ProducerBatch batch = iter.next(); //4.判断batch是否投递超时。默认消息投递过期时间是2 minutes if (batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now)) { //5.删除投递过期的batch iter.remove(); //6.如果 batch 没有最终的状态，就把batch加入到expiredBatches集合 if (!batch.isDone()) { expiredBatches.add(batch); } else { throw new IllegalStateException(batch.topicPartition + " batch created at " + batch.createdMs + " gets unexpected final state " + batch.finalState()); } } else { //更新下一个Batch的超时时间。 accumulator.maybeUpdateNextBatchExpiryTime(batch); break; } } if (partitionInFlightBatches.isEmpty()) { batchIt.remove(); } } } return expiredBatches; }

首先创建过期批次的集合expiredBatches，然后遍历inFlightBatches，拿到分区对应的List

后，再遍历List

，如果发现ProducerBatch存留的时间超过accumulator.getDeliveryTimeoutMs()就认为该批次过期了，DeliveryTimeoutMs的默认值是120s。如果发送的消息很多而且不太在意消息的延迟，可以考虑把这个参数适当增加。对于过期的批次，会从inFlightBatches里删除。

7. 处理过期的批次。

通过调用failBatch()方法实现对过期批次的处理，然后把过期的批次放入expiredBatches集合里。

//失败batch的处理 private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception, boolean adjustSequenceNumbers) { if (transactionManager != null) { transactionManager.handleFailedBatch(batch, exception, adjustSequenceNumbers); } this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount); //超时调用回调 if (batch.done(baseOffset, logAppendTime, exception)) { maybeRemoveAndDeallocateBatch(batch); } }

这个方法最关键的是代码13行里调用batch.done(）。具体是调用了batch里的回调方法。也就是说对于超时的批次不会发送给服务端了，而是直接执行回调方法完成请求。执行成功后删除批次并释放批次占用的空间。

8. 调用sendProduceRequests(batches, now）。我们看下这个方法：

private void sendProduceRequests(Map> collated, long now) { for (Map.Entry> entry : collated.entrySet()) sendProduceRequest(now, entry.getKey(), acks, requestTimeoutMs, entry.getValue()); }

本质上是轮询过滤完的批次集合去调用sendProduceRequest()方法，因为key是节点所以每调用一次就是向某个节点发送请求，而且是批量的发送因为value是List

。下面我们来介绍sendProduceRequest()方法，源码如下：

/** * Create a produce request from the given record batches * 把ProducerBatch类型转换成ClientRequest，并把clientRequest放到KafkaChannel的缓存里。 */ private void sendProduceRequest(long now, int destination, short acks, int timeout, List

batches) { if (batches.isEmpty()) return; //1.初始化两个集合，produceRecordsByPartition用于构建请求，recordsByPartition用于构建回调方法 Map produceRecordsByPartition = new HashMap<>(batches.size()); final Map recordsByPartition = new HashMap<>(batches.size()); ......忽略 //2.按分区填充produceRecordsByPartition和recordsByPartition两个集合。 for (ProducerBatch batch : batches) { TopicPartition tp = batch.topicPartition; MemoryRecords records = batch.records(); if (!records.hasMatchingMagic(minUsedMagic)) records = batch.records().downConvert(minUsedMagic, 0, time).records(); produceRecordsByPartition.put(tp, records); recordsByPartition.put(tp, batch); } ...... //3.创建requestBuilder对象。 ProduceRequest.Builder requestBuilder = ProduceRequest.Builder.forMagic(minUsedMagic, acks, timeout, produceRecordsByPartition, transactionalId); //4.创建回调 RequestCompletionHandler callback = response -> handleProduceResponse(response, recordsByPartition, time.milliseconds()); String = Integer.toString(destination); //5.创建clientRequest ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0, requestTimeoutMs, callback); //6.把clientRequest发送给NetworkClient，完成消息的预发送。 client.send(clientRequest, now); log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);

第一步，初始化两个集合，produceRecordsByPartition用于构建请求，recordsByPartition用于构建回调。

第二步，按分区填充这两个集合。

第三步，创建requestBuilder对象,。

第四步，创建callback对象。实际上是实例化回调接口RequestCompletionHandler，接口RequestCompletionHandler就一个方法onComplete()，callback对onComplete()的实现方式是用返回的response当参数调用handleProduceResponse()方法。消息发送的时候不会把callback方法发送到服务器，而是把callback放在NetworkClient里等待响应回来后根据响应调用callback。

第五步，创建clientRequest。用produceRecordsByPartition集合构建clientRequest，也就是说一个clientRequest对象里有多个消息批次。

第六步，把clientRequest交给NetworkClient，NetworkClient会调用相关方法把，clientRequest保存在kakfaChannel的send字段里，也就是是说预发送的最小单位是一个List

,也就是一个批次集合，而真正能不能一次就发送一个批次要看当时的网络环境。好，到这里完成消息的预发送。

发送消息设计到的类和方法给大家介绍完了，现在给大家介绍收到响应后Sender线程是如何处理的。

handleProduceResponse(ClientResponse response, Map batches, long now)

/** * Handle a produce response * 处理服务端发来的response。 */ private void handleProduceResponse(ClientResponse response, Map batches, long now) { RequestHeader requestHeader = response.requestHeader(); int correlationId = requestHeader.correlationId(); //连接失败 if (response.wasDisconnected()) { log.trace("Cancelled request with header {} due to node {} being disconnected", requestHeader, response.destination()); for (ProducerBatch batch : batches.values()) completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NETWORK_EXCEPTION), correlationId, now); //处理版本不匹配 } else if (response.versionMismatch() != null) { log.warn("Cancelled request {} due to a version mismatch with node {}", response, response.destination(), response.versionMismatch()); for (ProducerBatch batch : batches.values()) completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.UNSUPPORTED_VERSION), correlationId, now); //处理正常响应 } else { log.trace("Received produce response from node {} with correlation id {}", response.destination(), correlationId); // if we have a response, parse it // 有响应 if (response.hasResponse()) { ProduceResponse produceResponse = (ProduceResponse) response.responseBody(); for (Map.Entry entry : produceResponse.responses().entrySet()) { TopicPartition tp = entry.getKey(); ProduceResponse.PartitionResponse partResp = entry.getValue(); ProducerBatch batch = batches.get(tp); //调用completeBatch()方法处理。 completeBatch(batch, partResp, correlationId, now); } this.sensors.recordLatency(response.destination(), response.requestLatencyMs()); } else { //无响应，ack=0时的处理 for (ProducerBatch batch : batches.values()) { completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NONE), correlationId, now); } } } }

一个响应是某一个节点发给客户端的，一个节点每次向客户端发送的响应也是批量的，一个响应就有可能包含多个分区的响应信息。我们还要调用回调方法告诉生产者发送结果，而回调方法并没有在响应里，这要求我们要从请求的batch集合里拿到回调方法。于是，前面定义的recordsByPartition集合在处理响应的时候就有用了。

客户端接到响应后会根据响应的结果分情况处理：

处理方法是completeBatch()，只是根据具体情况配置的参数不同。

首先会处理两个失败的场景：如果连接失败会设定异常Errors.NETWORK_EXCEPTION，客户端和服务端api版本不匹配会设定异常Errors.UNSUPPORTED_VERSION。

正常响应也分两种情况，有返回值的响应和无返回值的响应。

1) 响应有返回值。根据响应集合里的分区找到recordsByPartition集合里对应的batch,然后把batch和响应信息当参数调用completeBatch()方法。

2）对应响应无返回值。的场景是，当发送时设定的ack=0时，生产端只管发送而不去管响应结果，所以不用考虑响应的返回值，也是调用completeBatch()方法。

好，最终还是要调用completeBatch()方法。我们下面看一下completeBatch()方法。

completeBatch():

方法源码如下：

private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, long correlationId, long now) { Errors error = response.error; //单调消息过长的消息，会把单条消息分成多个batch发送。 if (error == Errors.MESSAGE_TOO_LARGE && batch.recordCount > 1 && !batch.isDone() && (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 || batch.isCompressed())) { log.warn( "Got error produce response in correlation id {} on topic-partition {}, splitting and retrying ({} attempts left). Error: {}", correlationId, batch.topicPartition, this.retries - batch.attempts(), error); if (transactionManager != null) transactionManager.removeInFlightBatch(batch); this.accumulator.splitAndReenqueue(batch); maybeRemoveAndDeallocateBatch(batch); this.sensors.recordBatchSplit(); } else if (error != Errors.NONE) { //能否再次发送 if (canRetry(batch, response, now)) { 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); //再次入对 reenqueueBatch(batch, now); //重复发送 } else if (error == Errors.DUPLICATE_SEQUENCE_NUMBER) { //序列化重复的处理 completeBatch(batch, response); } else { final RuntimeException exception; //主题没有权限 if (error == Errors.TOPIC_AUTHORIZATION_FAILED) exception = new TopicAuthorizationException(Collections.singleton(batch.topicPartition.topic())); //集群没有权限 else if (error == Errors.CLUSTER_AUTHORIZATION_FAILED) exception = new ClusterAuthorizationException("The producer is not authorized to do idempotent sends"); //直接返回响应的错误信息 else exception = error.exception(response.errorMessage); failBatch(batch, response, exception, batch.attempts() < this.retries); } //无效的元数据异常 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); //其它元数据异常，直接返回响应的错误信息。 } else { log.warn("Received invalid metadata error in produce request on partition {} due to {}. Going " + "to request metadata update now", batch.topicPartition, error.exception(response.errorMessage).toString()); } metadata.requestUpdate(); } } else { //正常执行回调 completeBatch(batch, response); } // Unmute the completed partition. if (guaranteeMessageOrder) this.accumulator.unmutePartition(batch.topicPartition); }

首先，我们分析一下各种异常的处理：

1. 单调消息过长的异常。需要同时满足四个条件：

1）响应返回MESSAGE_TOO_LARGE异常。

2）batch中只有一条消息

3) batch没有完成。

4）消息的格式大于等于V2版本或消息是压缩的。

满足条件后进入if语句，因为消息太长了就分成多个批次发送，然后从集合中删除此batch，最后释放暂存器里此batch占用的空间。

2. 能再次入队等待发送的异常。只要满足canRetry()方法就可以再次尝试发送。canRetry()方法代码在下面，我给大家讲解一下：

canRetry()

private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response, long now) { return !batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now) && batch.attempts() < this.retries && !batch.isDone() && (transactionManager == null ? response.error.exception() instanceof RetriableException : transactionManager.canRetry(response, batch)); }

要同时满足四个条件，才能重新发送：

• 没有到投递的超时时间。
• batch重试次数没超过设定的次数。
• 批次没结束。
• 如果不被事务管理响应的异常属于可重试的异常，如果被事务管理调用事务管理器自定义的判断器判断是否能重试。

3. 序列化重复的异常。说明消息重复发送了，不用做任何处理。
4. 主题没有权限的异常。集群没有权限的异常。已经其他异常，统一调用方法failBatch（）处理。
5. 元数据的异常。包括找不到对应分区及其它的元数据异常。这两个异常除了打印对应的异常日志都会设定元数据需要更新的标识，因为响应已经报元数据异常了就应该及时更新元数据。
6. 响应没有异常。调用completeBatch(batch, response)完成响应处理。

接下来我们讨论处理正常响应的方法completeBatch(batch, response)：

completeBatch(batch, response)

private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response) { if (transactionManager != null) { transactionManager.handleCompletedBatch(batch, response); } //执行回调，并释放 accumular 的空间 if (batch.done(response.baseOffset, response.logAppendTime, null)) { maybeRemoveAndDeallocateBatch(batch); } }

这个方法就两个部分，调用batch的done()方法，然后从集合中删除batch并释放batch占用的内存。

好了，到目前为止，Sender类从发送消息到处理响应的主要方法都分析完了。现在总结一下今天学习的知识点。

总结：

Sender类是一个线程类，如果不主动关闭Sender会一直执行下去。我们讲解了Sender处理消息的流程，包括消息的发送和消息的响应处理。消息的发送还包括消息预发送和真正的网络发送。我们分析了Sender类的相关字段和方法。Sender类用到了很多其他组件，包括NetworkClient用于网络IO，RecordAccumulator用于获取要发送的消息。MetaData用于获取元数据。sendProducerData及相关方法用于消息的预发送。调用NetworkClient类的poll()方法用于真正的网络发送。completeBatch()方法包含了各种异常响应和正常响应的处理。handleProduceResponse()方法是回调方法的处理逻辑。