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的idboolean mark(long markSequence) throws InterruptedException;//放入id，返回在buffer中的Sequence（可以计算出数组下标），buffer满时会被阻塞（waiting for signal）long put(long id) throws InterruptedException;  
void setBufferSize(int bufferSize); //设置BufferSizevoid start(); //初始化buffervoid stop(); //关闭buff，重置putSequence，getSequence，清除buffer