Kafka-Clients源码学习:KafkaProducer篇

    private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long maxWaitMs) throws InterruptedException {
        // 将topic放入metadata中，如果已存在就刷新过期时间
        metadata.add(topic);
        // 从内存中fetch cluster元数据
        Cluster cluster = metadata.fetch();
        Integer partitionsCount = cluster.partitionCountForTopic(topic); // 找到了直接return
        if (partitionsCount != null && (partition == null || partition < partitionsCount))
            return new ClusterAndWaitTime(cluster, 0);

        long begin = time.milliseconds();
        long remainingWaitMs = maxWaitMs;
        long elapsed;
        // 标识需要进行元数据更新，并且等待至更新完成或抛出超时异常
        do {
            metadata.add(topic);
            int version = metadata.requestUpdate();
            sender.wakeup();
            try {
                metadata.awaitUpdate(version, remainingWaitMs);
            } catch (TimeoutException ex) {
                throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
            }
            //省略部分超时、权限异常检测代码
            ......
            partitionsCount = cluster.partitionCountForTopic(topic);
        } while (partitionsCount == null);

        if (partition != null && partition >= partitionsCount) {
            throw new KafkaException(
                    String.format("Invalid partition given with record: %d is not in the range [0...%d).", partition, partitionsCount));
        }

        return new ClusterAndWaitTime(cluster, elapsed);
    }    

    private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        TopicPartition tp = null;
        try {
            //省略部分
            //......
            byte[] serializedKey;
            try {
                //序列化key
                serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
            } catch (ClassCastException cce) {
                throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +
                        " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
                        " specified in key.serializer", cce);
            }
            byte[] serializedValue;
            try {
                //序列化value
                serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
            } catch (ClassCastException cce) {
                throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +
                        " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
                        " specified in value.serializer", cce);
            }
            
            Header[] headers = record.headers().toArray();
            int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(), compressionType, serializedKey, serializedValue, headers);
            //估算序列化后大小，确保不会超过max_request_size或者total_memory_size
            ensureValidRecordSize(serializedSize);
        }
        //省略部分
        //......
    }

    public RecordAppendResult append(TopicPartition tp,
                                     long timestamp,
                                     byte[] key,
                                     byte[] value,
                                     Header[] headers,
                                     Callback callback,
                                     long maxTimeToBlock) throws InterruptedException {
        ByteBuffer buffer = null;
        if (headers == null) headers = Record.EMPTY_HEADERS;
        try {
            //根据TopicPartition获得对应双端队列dq，没有则创建
            Deque<ProducerBatch> dq = getOrCreateDeque(tp);
            synchronized (dq) {
                //尝试向已有dq中的last batch已写入此条record
                RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
                if (appendResult != null)
                    return appendResult;
            }

            // dq中的last batch无法再添加record，于是申请一个新的batch，并且添加到dq的last
            // 从BufferPool中拿出一个buffer
            buffer = free.allocate(size, maxTimeToBlock);
            synchronized (dq) {
                RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
                if (appendResult != null) {
                    //两次获取dq锁的过程中可能有其它线程创建了可用的batch，如果tryAppend成功则直接返回
                    return appendResult;
                }

                MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic);
                ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds());
                //将record append进刚刚新建的batch
                FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()));
                // 将申请的batch放入dp last
                dq.addLast(batch);
                incomplete.add(batch);

                // 避免在finally模块中释放这次申请的buffer
                buffer = null;

                return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true);
            }
        } finally {
            if (buffer != null)
                // 如果使用了其它线程申请的buffer则释放此线程申请的buffer
                free.deallocate(buffer);
        }
    }

    private long sendProducerData(long now) {
        //获取本地元数据
        Cluster cluster = metadata.fetch();

        // 获取存在待发送分区对应的node
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

        // 如果本地元数据中没找到对应leader信息，强制元数据更新
        if (!result.unknownLeaderTopics.isEmpty()) {
            for (String topic : result.unknownLeaderTopics)
                this.metadata.add(topic);
            this.metadata.requestUpdate();
        }

        // 删除没有准备好发送的node
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            // 检查是否能向此node发送消息（这里会保证等待同一node的上次request send发送完成）
            if (!this.client.ready(node, now)) {
                iter.remove();
                // 获得disconnected状态的node剩余backoff时间 
                notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
            }
        }

        // 从Accumulator中获取准备发送的batchs
        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes,
                this.maxRequestSize, now);
        if (guaranteeMessageOrder) {
            // 如果max.inflight.requests = 1，则将待发送的partition锁住
            for (List<ProducerBatch> batchList : batches.values()) {
                for (ProducerBatch batch : batchList)
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }
        //从accumulator中拿出timeout的batch(max.inflight.requests > 1的情况下过期不保回调顺序)
        List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(this.requestTimeout, now);
        for (ProducerBatch expiredBatch : expiredBatches) {
            //回调timeout异常
            failBatch(expiredBatch, -1, NO_TIMESTAMP, expiredBatch.timeoutException(), false);
            //......
        }

        // 如果存在可发送数据并且准备好发送的node，会以0的timeout来返回pollTimeout，否则此timeout会由当前无法发送消息的nodes决定（比如 lingering, backing off）。总的来说：
        // 1. 如果存在partition的数据准备好发送，pollTimeout=0
        // 2. 另外如果存在partition有数据写入accumulator但是没有准备好发送，pollTimeout=min{当前时间与linger过期时间差值(重发的batch考虑retryBackoff差值), 当前时间与disconnected node的reconnectBackoffMs的时间差值}
        long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
        if (!result.readyNodes.isEmpty()) {
            pollTimeout = 0;
        }
        //将batchs构建成request发送到缓存
        sendProduceRequests(batches, now);

        return pollTimeout;
    }

    private void sendProduceRequests(Map<Integer, List<ProducerBatch>> collated, long now) {
        // 遍历batches，以node为单位发送request请求
        for (Map.Entry<Integer, List<ProducerBatch>> entry : collated.entrySet())
            sendProduceRequest(now, entry.getKey(), acks, requestTimeout, entry.getValue());
    }


    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());

        //......
        for (ProducerBatch batch : batches) {
            TopicPartition tp = batch.topicPartition;
            MemoryRecords records = batch.records();
            //......
            produceRecordsByPartition.put(tp, records);
            recordsByPartition.put(tp, batch);
        }

        // 如果producer开启了事务属性则获取transactionId
        String transactionalId = null;
        if (transactionManager != null && transactionManager.isTransactional()) {
            transactionalId = transactionManager.transactionalId();
        }
        // 用produceRecordsByPartition构建requestBuilder
        ProduceRequest.Builder requestBuilder = ProduceRequest.Builder.forMagic(minUsedMagic, acks, timeout,
                produceRecordsByPartition, transactionalId);
        // 用recordsByPartition构建回调类，client.poll收到response后会调用callback.onComplete方法。
        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发送消息
        client.send(clientRequest, now);
        log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
    }

public List<ClientResponse> poll(long timeout, long now) {
    if (!abortedSends.isEmpty()) { // 如果存在abortedSends，直接处理，不经过selector的poll
        List<ClientResponse> responses = new ArrayList<>();
        handleAbortedSends(responses);
        completeResponses(responses);
        return responses;
    }
 
    //判断是否要更新元数据，如果需要的话发送更新请求（会选择inflight请求数量最少的ready node发送请求）
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
        // 完成一次selector poll，最大阻塞时间为timeout，metadataTimeout，requestTimeoutMs的最小值
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }
 
    long updatedNow = this.time.milliseconds();
    // 初始化response list
    List<ClientResponse> responses = new ArrayList<>();
    // 如果是不等待回复的请求则直接生成response放入list
    handleCompletedSends(responses, updatedNow);
    // 根据inFlightRequests的发送顺序生成已完成回复的response放入list
    handleCompletedReceives(responses, updatedNow);
    // 处理断开连接的请求，生成response放入list
    handleDisconnections(responses, updatedNow);
    // 更新connectionStates中node的连接状态
    handleConnections();
    // 需要的话发送api version获取请求
    handleInitiateApiVersionRequests(updatedNow);
    // 处理超时请求，生成response放入list
    handleTimedOutRequests(responses, updatedNow);
    // 完成所有response的处理
    completeResponses(responses);
 
    return responses;
}
 
private void completeResponses(List<ClientResponse> responses) {
    for (ClientResponse response : responses) {
        try {
            // 调用response中的callback进行处理，producer端可以参考Sender类的handleProduceResponse方法
            response.onComplete();
        } catch (Exception e) {
            log.error("Uncaught error in request completion:", e);
        }
    }
}

    private void handleProduceResponse(ClientResponse response, Map<TopicPartition, ProducerBatch> batches, long now) {
        //......
        if (response.wasDisconnected()) { 
            // 处理连接断开batches
            for (ProducerBatch batch : batches.values())
                completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NETWORK_EXCEPTION), correlationId, now);
        } else if (response.versionMismatch() != null) { 
            // 处理协议版本不匹配
            for (ProducerBatch batch : batches.values())
                completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.UNSUPPORTED_VERSION), correlationId, now);
        } else {
            // 如果response有内容，则解析后completeBatch
            if (response.hasResponse()) {
                ProduceResponse produceResponse = (ProduceResponse) response.responseBody();
                for (Map.Entry<TopicPartition, ProduceResponse.PartitionResponse> entry : produceResponse.responses().entrySet()) {
                    TopicPartition tp = entry.getKey();
                    ProduceResponse.PartitionResponse partResp = entry.getValue();
                    ProducerBatch batch = batches.get(tp);
                    completeBatch(batch, partResp, correlationId, now);
                }
            } else {
                // 如果没有response，即ack = 0 时会走此处逻辑，直接completeBatch
                for (ProducerBatch batch : batches.values()) {
                    completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NONE), correlationId, now);
                }
            }
        }
    }

//获得buffer中的下一个元素，在buffer为空或者待取出id没有被标记过的时候阻塞（waiting for signal）
long get() throws InterruptedException;
//标记传入Sequence的id
boolean mark(long markSequence) throws InterruptedException;
//放入id，返回在buffer中的Sequence（可以计算出数组下标），buffer满时会被阻塞（waiting for signal）
long put(long id) throws InterruptedException;  

void setBufferSize(int bufferSize); //设置BufferSize
void start(); //初始化buffer
void stop(); //关闭buff，重置putSequence，getSequence，清除buffer