-     本次目标     -

上一小节我们剖析了如果响应消息里带有异常KafkaProducer是如何处理的。接着这一讲我们剖析KafkaProducer端如果监测到有消息超时了但是还没被发送出去，KafkaProducer端是如何处理的？大家是否还记得消息被发送之前首先会在RecordAccumulator里面被封装成批次，装在batchs里面。我们这儿说的就是，如果消息在batches里面超时了，KafkaProducer是如何处理这些消息的? 

-     源码剖析     -

我们再次回到Sender线程的run方法：

 void run(long now) {
        .....//TODO 步骤六：放弃超时的batch//在这个里面我们看一下，对于超时的batch是如何处理的。
        List expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);// update sensorsfor (RecordBatch expiredBatch : expiredBatches)this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);
        ......this.client.poll(pollTimeout, now);
    }

查看核心方法：

 public List abortExpiredBatches(int requestTimeout, long now) {
        List expiredBatches = new ArrayList<>();int count = 0;/**
         * 遍历batchs里面的所有partition
         */for (Map.Entry> entry : this.batches.entrySet()) {
            Deque dq = entry.getValue();
            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.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();//获取partition队列里面的所有batch
                    Iterator batchIterator = dq.iterator();//遍历每个batchwhile (batchIterator.hasNext()) {
                        RecordBatch batch = batchIterator.next();
                        boolean isFull = batch != lastBatch || batch.records.isFull();// TODO 对每个batch检测是否超时if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) {//把超时的batch添加到超时的数据结构作为返回值返回去。
                            expiredBatches.add(batch);
                            count++;
                            batchIterator.remove();//如果超时了释放内存
                            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;
    }

我们重点看一下判断是否超时的方法：

/**
     * 这个方法是判断batch是否超时的
     * @param requestTimeoutMs 超时的时间，默认是30秒
     * @param retryBackoffMs
     * @param now
     * @param lingerMs
     * @param isFull
     * @return
     */public boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) {
        boolean expire = false;
        String errorMessage = null;/**
         *  (1)如果当前时间 减去 这个批次加入队列的时候已经超过30秒了，那么就已经超时了。
         */if (!this.inRetry() && isFull && requestTimeoutMs this.lastAppendTime)) {
            expire = true;
            errorMessage = (now - this.lastAppendTime) + " ms has passed since last append";/**
             * (2)当前时间减去创建批次的时间和批次应该发送的时间间隔大于30秒
             * 那么说明已经超时了
             */
        } else if (!this.inRetry() && requestTimeoutMs this.createdMs + lingerMs))) {
            expire = true;
            errorMessage = (now - (this.createdMs + lingerMs)) + " ms has passed since batch creation plus linger time";/**
             * (3)如果当前batch是重试的，并且当前时间减去上一次重试的时间和重试时间间隔 大于30秒
             * 那么说明已经超时了。
             */
        } else if (this.inRetry() && requestTimeoutMs this.lastAttemptMs + retryBackoffMs))) {
            expire = true;
            errorMessage = (now - (this.lastAttemptMs + retryBackoffMs)) + " ms has passed since last attempt plus backoff time";
        }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;
    }

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);// execute callbacks//一个thunks就代表一个消息for (int i = 0; i this.thunks.size(); i++) {try {
                Thunk thunk = this.thunks.get(i);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 {//如果响应里有异常，那么就把异常也传给回调函数。
                    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);
    }

综上所述，我们发现如果KafkaProducer端检测到batchs超时了，那么就会释放对应内存，最终也会调用消息的回调函数，把超时异常带过去。

-     总结     -

这一小节，我们分析了一个批次要是长时间没被发送出去KafkaProducer端是如何处理的。其实到目前为止我们的KafkaProducer端的整个流程就剖析得差不多了，下一讲剖析KafkaPrducer端最后一个知识点，当一个batch被发送出去以后长时间没有接受到响应，KafkaProducer端是如何处理的？大家加油！！！-   关注“大数据观止”   -