Kafka 源码解析之 Producer 单 Partition 顺序性实现及配置说明（五） | Matt's Blog

1     
2     
3     
4     
5     
6     
7     
8     
9     

private final Set<TopicPartition> muted;     
     
public void mutePartition(TopicPartition tp) {     
    muted.add(tp);     
}     
     
public void unmutePartition(TopicPartition tp) {     
    muted.remove(tp);     
}     

1     
2     
3     
4     
5     
6     
7     
8     
9     
10     
11     
12     
13     
14     
15     
16     
17     
18     
19     
20     
21     
22     
23     
24     
25     
26     
27     
28     
29     
30     
31     
32     
33     
34     
35     
36     
37     
38     
39     
40     
41     
42     

public ReadyCheckResult ready(Cluster cluster, long nowMs) {     
    Set<Node> readyNodes = new HashSet<>();     
    long nextReadyCheckDelayMs = Long.MAX_VALUE;     
    Set<String> unknownLeaderTopics = new HashSet<>();     
     
    boolean exhausted = this.free.queued() > 0;     
    for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {     
        TopicPartition part = entry.getKey();     
        Deque<RecordBatch> deque = entry.getValue();     
     
        Node leader = cluster.leaderFor(part);     
        synchronized (deque) {     
            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)) {//note: part 如果 mute 就不会遍历     
                RecordBatch batch = deque.peekFirst();     
                if (batch != null) {     
                    boolean backingOff = batch.attempts > 0 && batch.lastAttemptMs + retryBackoffMs > nowMs;     
                    //note: 是否是在重试     
                    long waitedTimeMs = nowMs - batch.lastAttemptMs;     
                    long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;     
                    long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);     
                    boolean full = deque.size() > 1 || batch.isFull(); //note: batch 满了     
                    boolean expired = waitedTimeMs >= timeToWaitMs; //note: batch 超时     
                    boolean sendable = full || expired || exhausted || closed || flushInProgress();     
                    if (sendable && !backingOff) {     
                        readyNodes.add(leader);// note: 将可以发送的 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);     
}     

1     
2     
3     
4     
5     
6     
7     
8     
9     
10     
11     
12     
13     
14     
15     
16     
17     
18     
19     
20     
21     
22     
23     
24     
25     
26     
27     
28     
29     
30     
31     
32     
33     
34     
35     
36     
37     
38     
39     
40     
41     
42     
43     
44     
45     
46     
47     
48     
49     
50     
51     

//note: 返回该 node 对应的可以发送的 RecordBatch 的 batches,并从 queue 中移除（最大的大小为maxSize,超过的话,下次再发送）     
public Map<Integer, List<RecordBatch>> drain(Cluster cluster,     
                                             Set<Node> nodes,     
                                             int maxSize,     
                                             long now) {     
    if (nodes.isEmpty())     
        return Collections.emptyMap();     
     
    Map<Integer, List<RecordBatch>> batches = new HashMap<>();     
    for (Node node : nodes) {     
        int size = 0;     
        List<PartitionInfo> parts = cluster.partitionsForNode(node.id());     
        List<RecordBatch> ready = new ArrayList<>();     
        /* to make starvation less likely this loop doesn't start at 0 */     
        int start = drainIndex = drainIndex % parts.size();     
        do {     
            PartitionInfo part = parts.get(drainIndex);     
            TopicPartition tp = new TopicPartition(part.topic(), part.partition());     
            // Only proceed if the partition has no in-flight batches.     
            if (!muted.contains(tp)) {//note: 被 mute 的 tp 依然不会被遍历     
                Deque<RecordBatch> deque = getDeque(new TopicPartition(part.topic(), part.partition()));     
                if (deque != null) {     
                    synchronized (deque) {     
                        RecordBatch first = deque.peekFirst();     
                        if (first != null) {     
                            boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;     
                            // Only drain the batch if it is not during backoff period.     
                            if (!backoff) {     
                                if (size + first.sizeInBytes() > maxSize && !ready.isEmpty()) {     
                                    // there is a rare case that a single batch size is larger than the request size due     
                                    // to compression; in this case we will still eventually send this batch in a single     
                                    // request     
                                    break;     
                                } else {     
                                    RecordBatch batch = deque.pollFirst();     
                                    batch.close();     
                                    size += batch.sizeInBytes();     
                                    ready.add(batch);     
                                    batch.drainedMs = now;     
                                }     
                            }     
                        }     
                    }     
                }     
            }     
            this.drainIndex = (this.drainIndex + 1) % parts.size();     
        } while (start != drainIndex);     
        batches.put(node.id(), ready);     
    }     
    return batches;     
}     

1     
2     
3     
4     
5     
6     
7     
8     
9     
10     
11     
12     
13     
14     
15     
16     
17     
18     
19     
20     
21     
22     
23     
24     
25     
26     
27     
28     
29     
30     
31     
32     
33     
34     
35     

// from KafkaProducer     
this.sender = new Sender(client,     
                         this.metadata,     
                         his.accumulator,     
                         config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) == 1,     
                         config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),     
                         (short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),     
                         config.getInt(ProducerConfig.RETRIES_CONFIG),     
                         this.metrics,     
                         Time.SYSTEM,     
                         this.requestTimeoutMs);//NOTE: Sender 实例,发送请求的后台线程     
     
// from Sender     
public Sender(KafkaClient client,     
              Metadata metadata,     
              RecordAccumulator accumulator,     
              boolean guaranteeMessageOrder,     
              int maxRequestSize,     
              short acks,     
              int retries,     
              Metrics metrics,     
              Time time,     
              int requestTimeout) {     
        this.client = client;     
        this.accumulator = accumulator;     
        this.metadata = metadata;     
        this.guaranteeMessageOrder = guaranteeMessageOrder;     
        this.maxRequestSize = maxRequestSize;     
        this.running = true; //note: 默认为 true     
        this.acks = acks;     
        this.retries = retries;     
        this.time = time;     
        this.sensors = new SenderMetrics(metrics);     
        this.requestTimeout = requestTimeout;     
}     

1     
2     
3     
4     
5     
6     
7     
8     
9     

//from Sender     
//note: max.in.flight.requests.per.connection 设置为1时会保证     
if (guaranteeMessageOrder) {     
    // Mute all the partitions draine     
    for (List<RecordBatch> batchList : batches.values()) {     
         for (RecordBatch batch : batchList)     
             this.accumulator.mutePartition(batch.topicPartition);     
    }     
}