Kafka成长记3：Producer 元数据拉取源码原理（上）

上一节我们分析了Producer的核心组件，我们得到了一张关键的组件图。你还记得么？

简单概括下上面的图就是：

创建了Metadata组件，内部通过Cluster维护元数据

初始化了发送消息的内存缓冲器RecordAccumulator

创建了NetworkClient，内部最重要的是创建了NIO的Selector组件

启动了一个Sender线程，Sender引用了上面的所有组件，开始执行run方法。

图的最下方可以看到，上一节截止到了run方法的执行，这一节我们首先会看看run方法核心脉络做了什么。接着分析下Producer第一个核心流程：元数据拉取的源码原理。

让我们开始吧！

Sender的run方法在做什么？

这一节我们就继续分析下，sender的run方法开始执行会做什么。

 public void run() {
        log.debug("Starting Kafka producer I/O thread.");

        // main loop, runs until close is called
        while (running) {
            try {
                run(time.milliseconds());
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }

        log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");

        // okay we stopped accepting requests but there may still be
        // requests in the accumulator or waiting for acknowledgment,
        // wait until these are completed.
        while (!forceClose && (this.accumulator.hasUnsent() || this.client.inFlightRequestCount() > 0)) {
            try {
                run(time.milliseconds());
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }
        if (forceClose) {
            // We need to fail all the incomplete batches and wake up the threads waiting on
            // the futures.
            this.accumulator.abortIncompleteBatches();
        }
        try {
            this.client.close();
        } catch (Exception e) {
            log.error("Failed to close network client", e);
        }

        log.debug("Shutdown of Kafka producer I/O thread has completed.");
    }

这个run方法的核心脉络很简单。主要就是2个while循环+线程的close，而2个while循环，他们都调用了run(long time)的这个方法。

通过注释你可以看到，第二个while是处理特殊情况的，当第一个while退出后，还有未发送的请求，需要第二个while循环处理完成，才会关闭线程。

整体脉络如下图所示:

接着其实就该看下run方法主要在干什么了？

   /**
     * Run a single iteration of sending
     * 
     * @param now
     *            The current POSIX time in milliseconds
     */
    void run(long now) {
        Cluster cluster = metadata.fetch();
        // get the list of partitions with data ready to send
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

        // if there are any partitions whose leaders are not known yet, force metadata update
        if (result.unknownLeadersExist)
            this.metadata.requestUpdate();

        // remove any nodes we aren"t ready to send to
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            if (!this.client.ready(node, now)) {
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
            }
        }

        // create produce requests
        Map<Integer, List<RecordBatch>> batches = this.accumulator.drain(cluster,
                                                                         result.readyNodes,
                                                                         this.maxRequestSize,
                                                                         now);
        if (guaranteeMessageOrder) {
            // Mute all the partitions drained
            for (List<RecordBatch> batchList : batches.values()) {
                for (RecordBatch batch : batchList)
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }

        List<RecordBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);
        // update sensors
        for (RecordBatch expiredBatch : expiredBatches)
            this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);

        sensors.updateProduceRequestMetrics(batches);
        List<ClientRequest> requests = createProduceRequests(batches, now);
        // If we have any nodes that are ready to send + have sendable data, poll with 0 timeout so this can immediately
        // loop and try sending more data. Otherwise, the timeout is determined by nodes that have partitions with data
        // that isn"t yet sendable (e.g. lingering, backing off). Note that this specifically does not include nodes
        // with sendable data that aren"t ready to send since they would cause busy looping.
        long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
        if (result.readyNodes.size() > 0) {
            log.trace("Nodes with data ready to send: {}", result.readyNodes);
            log.trace("Created {} produce requests: {}", requests.size(), requests);
            pollTimeout = 0;
        }
        for (ClientRequest request : requests)
            client.send(request, now);

        // if some partitions are already ready to be sent, the select time would be 0;
        // otherwise if some partition already has some data accumulated but not ready yet,
        // the select time will be the time difference between now and its linger expiry time;
        // otherwise the select time will be the time difference between now and the metadata expiry time;
        this.client.poll(pollTimeout, now);
    }

上面的代码，你如果第一次看，你肯定会觉得，这个脉络非常不清晰，不知道重点在哪里。不过还好有些注释，你能大体猜到他在干嘛。

accumulator的ready，networkclient的ready、networkclient的send、networkclient的poll

这些好像是在准备内存区域、准备网络连接的node节点、发送数据、拉取响应结果的意思。

可是如果你猜不到，该怎么办呢？

这时候就可以祭出debug这个杀器了。由于是producer，我们可以在Hellowolrd的这个客户端打断点，一步一步看下。

当你对run方法一步一步打了断点之后你会发现：

accumulator的ready，networkclient的ready、networkclient的send 这些的逻辑几乎都没有执行，全部都是初始化空对象，或者方法内部直接return。

直接一路执行到了client.poll方法。如下图所示：

那么，你可以得出一个结论，while第一次循环这个run方法的核心逻辑，其实只有一句话：

client.poll(pollTimeout, now)

整体脉络如下所示：

看来接下来，这个NetworkClient的poll方法，就是关键中的关键了：

    /**
     * Do actual reads and writes to sockets.
     * 对套接字进行实际读取和写入
     *
     * @param timeout The maximum amount of time to wait (in ms) for responses if there are none immediately,
     *                must be non-negative. The actual timeout will be the minimum of timeout, request timeout and
     *                metadata timeout
     * @param now The current time in milliseconds
     * @return The list of responses received
     */
    @Override
    public List<ClientResponse> poll(long timeout, long now) {
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // process completed actions
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleTimedOutRequests(responses, updatedNow);

        // invoke callbacks
        for (ClientResponse response : responses) {
            if (response.request().hasCallback()) {
                try {
                    response.request().callback().onComplete(response);
                } catch (Exception e) {
                    log.error("Uncaught error in request completion:", e);
                }
            }
        }

        return responses;
    }

这个方法的脉络就清晰多了,通过方法名和注释，我们几乎可以猜出他的一些作用主要有：

1）注释说：对套接字进行实际读取和写入

2）metadataUpdater.maybeUpdate()，你还记得NetworkClient的组件DefaultMetadataUpdater么，方法名意思是可能进行元数据更新。这个好像很关键的样子

3）接着执行了Selector的poll方法，这个是NetworkClient的另一个组件Selector，还记得么？它底层封装了原生的NIO Selector。这个方法应该也比较关键。

4）后续对response执行了一系列的方法，从名字上看， handleCompletedSends 处理完成发送的请求、handleCompletedReceives处理完成接受的请求、handleDisconnections处理断开连接的请求、handleConnections处理连接成功的请求、处理超时的请求handleTimedOutRequests。根据不同情况有不同的处理。

5）最后还有一个response的相关的回调处理，如果注册了回调函数，会执行下。这个应该不是很关键的逻辑

也就是简单的说就是NetworkClient执行poll方法，主要通过selector处理请求的读取和写入，对响应结果做不同的处理而已。

如下图所示：

到这里其实我们基本摸清出了run方法主要在做的一件事情了，由于是第一次循环，之前的accumulator的ready，networkclient的ready、networkclient的send 什么都没做，第一次while循环run方法核心执行的是networkclient.poll方法。而poll方法的主要逻辑就是上面图中所示的了。

maybeUpdate可能在在拉取元数据？

刚才我们分析到，poll方法首先执行的是DefaultMetadataUpdater的maybeUpdate方法，它是可能更新的意思。我们来一起看下他的逻辑吧。

        public long maybeUpdate(long now) {
            // should we update our metadata?
            long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
            long timeToNextReconnectAttempt = Math.max(this.lastNoNodeAvailableMs + metadata.refreshBackoff() - now, 0);
            long waitForMetadataFetch = this.metadataFetchInProgress ? Integer.MAX_VALUE : 0;
            // if there is no node available to connect, back off refreshing metadata
            long metadataTimeout = Math.max(Math.max(timeToNextMetadataUpdate, timeToNextReconnectAttempt),
                    waitForMetadataFetch);

            if (metadataTimeout == 0) {
                // Beware that the behavior of this method and the computation of timeouts for poll() are
                // highly dependent on the behavior of leastLoadedNode.
                Node node = leastLoadedNode(now);
                maybeUpdate(now, node);
            }

            return metadataTimeout;
        }   

   /**
     * The next time to update the cluster info is the maximum of the time the current info will expire and the time the
     * current info can be updated (i.e. backoff time has elapsed); If an update has been request then the expiry time
     * is now
     */
    public synchronized long timeToNextUpdate(long nowMs) {
        long timeToExpire = needUpdate ? 0 : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0);
        long timeToAllowUpdate = this.lastRefreshMs + this.refreshBackoffMs - nowMs;
        return Math.max(timeToExpire, timeToAllowUpdate);
    }

原来这里有一个时间的判断，当判断满足才会执行maybeUpdate。

这个时间计算好像比较复杂，但是大体可以看出来，metadataTimeout是根据三个时间综合判断出来的，如果是0才会执行真正的maybeUpdate()。

像这种时候，我们可以直接在metadataTimeout这里打一个断点，看下它的值是如何计算的，比如下图：

你会发现，当第一次执行while循环，执行到poll方法，执行到这个maybeUpdate的时候，决定metadataTimeout的3个值，有两个是0，其中一个是非0，是一个299720的值。最终导致metadataTimeout也是非0，是299720。

也就是说，第一次while循环不会执行maybeUpdate的任何逻辑。

那么接着向下执行 Selector的poll()方法。

   /**
     * Do whatever I/O can be done on each connection without blocking. This includes completing connections, completing
     * disconnections, initiating new sends, or making progress on in-progress sends or receives.
     * 在不阻塞的情况下，在每个连接上做任何可以做的 I/O。这包括完成连接完成、断开连接，启动新的发送，或在进行中的发送或接收请求
     */
    @Override
    public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");

        clear();

        if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty())
            timeout = 0;

        /* check ready keys */
        long startSelect = time.nanoseconds();
        //这个方法是NIO底层Selector.select()，会阻塞监听
        int readyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        currentTimeNanos = endSelect;
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

        //如果监听到有操作的SelectionKeys,也就是readyKeys>0< 会执行一些操作
        if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
            pollSelectionKeys(this.nioSelector.selectedKeys(), false);
            pollSelectionKeys(immediatelyConnectedKeys, true);
        }

        addToCompletedReceives();

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
        maybeCloseOldestConnection();
    }

    private int select(long ms) throws IOException {
        if (ms < 0L)
            throw new IllegalArgumentException("timeout should be >= 0");

        if (ms == 0L)
            return this.nioSelector.selectNow();
        else
            return this.nioSelector.select(ms);
    }

上面的脉络主要是2步：

1）select(timeout): NIO底层selector.select()，会阻塞监听

2）pollSelectionKeys(): 监听到有操作的SelectionKeys，做了一些操作

也就是说，最终，Sender线程的run方法，第一次while循环执行poll方法，最后什么都没干，会被selector.select()阻塞住。

如下图所示：

new KafkaProducer之后

分析完了run方法的执行 ，我们分析的KafkaProducerHelloWorld第一步new KafkaProducer()基本就完成了。

大家经历了一节半的时间，终于分析清楚了KafkaProducer创建的原理。不不知道你对Kafka的Producer是不是有了更深的理解了。

分析了new KafkaProducer()之后呢？

我们继续接着KafkaProducerHelloWorld往下分析，你还记得KafkaProducerHelloWorld的代码么？

public class KafkaProducerHelloWorld {

	public static void main(String[] args) throws Exception {
		//配置Kafka的一些参数
		Properties props = new Properties();
		props.put("bootstrap.servers", "mengfanmao.org:9092");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		// 创建一个Producer实例
		KafkaProducer<String, String> producer = new KafkaProducer<>(props);

		// 封装一条消息
		ProducerRecord<String, String> record = new ProducerRecord<>(
				"test-topic", "test-key", "test-value");

		// 同步方式发送消息，会阻塞在这里，直到发送完成
		// producer.send(record).get();

		// 异步方式发送消息，不阻塞，设置一个监听回调函数即可
		producer.send(record, new Callback() {
			@Override
			public void onCompletion(RecordMetadata metadata, Exception exception) {
				if(exception == null) {
					System.out.println("消息发送成功");
				} else {
					exception.printStackTrace();
				}
			}
		});

		Thread.sleep(5 * 1000);

		// 退出producer
		producer.close();
	}

KafkaProducerHelloWorld主要就3步：

1）new KafkaProducer 这个我们已经分析完了，主要分析了配置文件的解析、各个组件是什么、有什么，还有就是刚才分析的run线程第一次循环到底执行了什么。

2） new ProducerRecord 创建待发送的消息

3） producer.send() 发送消息

首先创建待发送的消息：

ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");

public ProducerRecord(String topic, K key, V value) {
    this(topic, null, null, key, value);
}
    /**
     * Creates a record with a specified timestamp to be sent to a specified topic and partition
     * 创建具有指定时间戳的记录以发送到指定主题和分区
     * @param topic The topic the record will be appended to
     * @param partition The partition to which the record should be sent
     * @param timestamp The timestamp of the record
     * @param key The key that will be included in the record
     * @param value The record contents
     */
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
        if (topic == null)
            throw new IllegalArgumentException("Topic cannot be null");
        if (timestamp != null && timestamp < 0)
            throw new IllegalArgumentException("Invalid timestamp " + timestamp);
        this.topic = topic;
        this.partition = partition;
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
    }

我们之前提过，Record表示了一条消息的抽象封装。这个ProducerRecord其实就表示了一条消息。

从构造函数的注释可以看出来，ProducerRecord可以指定往哪个topic，哪一个分区partition，并且消息可以设置一个时间戳。分区和时间戳默认可以不指定

其实看这块源码，我们主要得到的信息就是这些了，这些都比较简单。就不画图了。

发送消息时的元数据拉取触发

当Producer和Record都创建好了之后，可以用同步或者异步的方式发送消息。

// 同步方式发送消息，会阻塞在这里，直到发送完成
// producer.send(record).get();

// 异步方式发送消息，不阻塞，设置一个监听回调函数即可
producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if(exception == null) {
            System.out.println("消息发送成功");
        } else {
            exception.printStackTrace();
        }
    }
});

    //同步发送
    @Override
    public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
        return send(record, null);
    }
    //异步发送
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // intercept the record, which can be potentially modified; this method does not throw exceptions
        ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
        return doSend(interceptedRecord, callback);
    }

同步和异步的整个发送逻辑如下图所示：

从上图你会发现，但是无论同步发送还是异步底层都会调用同一个方法doSend()。区别就是有没有callBack回调函数而已，他们还都在调用前注册一些拦截器，这里我们抓大放小下，我们重点还是关注doSend方法。

doSend方法如下：

/**
 * Implementation of asynchronously send a record to a topic. Equivalent to <code>send(record, null)</code>.
 * See {@link #send(ProducerRecord, Callback)} for details.
 */
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    TopicPartition tp = null;
    try {
        // first make sure the metadata for the topic is available
        long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
        long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - waitedOnMetadataMs);
        byte[] serializedKey;
        try {
            serializedKey = keySerializer.serialize(record.topic(), 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");
        }
        byte[] serializedValue;
        try {
            serializedValue = valueSerializer.serialize(record.topic(), 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");
        }
        int partition = partition(record, serializedKey, serializedValue, metadata.fetch());
        int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
        ensureValidRecordSize(serializedSize);
        tp = new TopicPartition(record.topic(), partition);
        long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
        log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
        // producer callback will make sure to call both "callback" and interceptor callback
        Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
        RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
        if (result.batchIsFull || result.newBatchCreated) {
            log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
            this.sender.wakeup();
        }
        return result.future;
        // handling exceptions and record the errors;
        // for API exceptions return them in the future,
        // for other exceptions throw directly
    } catch (ApiException e) {
        log.debug("Exception occurred during message send:", e);
        if (callback != null)
            callback.onCompletion(null, e);
        this.errors.record();
        if (this.interceptors != null)
            this.interceptors.onSendError(record, tp, e);
        return new FutureFailure(e);
    } catch (InterruptedException e) {
        this.errors.record();
        if (this.interceptors != null)
            this.interceptors.onSendError(record, tp, e);
        throw new InterruptException(e);
    } catch (BufferExhaustedException e) {
        this.errors.record();
        this.metrics.sensor("buffer-exhausted-records").record();
        if (this.interceptors != null)
            this.interceptors.onSendError(record, tp, e);
        throw e;
    } catch (KafkaException e) {
        this.errors.record();
        if (this.interceptors != null)
            this.interceptors.onSendError(record, tp, e);
        throw e;
    } catch (Exception e) {
        // we notify interceptor about all exceptions, since onSend is called before anything else in this method
        if (this.interceptors != null)
            this.interceptors.onSendError(record, tp, e);
        throw e;
    }
}

这个方法的脉络虽然比较长，但是脉络还是比较清晰，主要先执行了：

1）waitOnMetadata 应该是等待元数据拉取

2）keySerializer.serialize和valueSerializer.serialize，很明显就是将Record序列化成byte字节数组

3）通过partition进行路由分区，按照一定路由策略选择Topic下的某个分区

4）accumulator.append将消息放入缓冲器中

5）唤醒Sender线程的selector.select()的阻塞，开始处理内存缓冲器中的数据。

用图来表示如下所示：

这两节我们重点分析元数据拉取的这个场景的源码原理。

所以这里我们着重先看下步骤1 ，之后的4步我们之后会分析到的。

waitOnMetadata 如何等待元数据拉取的?

既然send的第一步是执行waitOnMetadata方法，首先看下它的代码：

 /**
     * Wait for cluster metadata including partitions for the given topic to be available.
     * @param topic The topic we want metadata for
     * @param maxWaitMs The maximum time in ms for waiting on the metadata
     * @return The amount of time we waited in ms
     */
    private long waitOnMetadata(String topic, long maxWaitMs) throws InterruptedException {
        // add topic to metadata topic list if it is not there already.
        if (!this.metadata.containsTopic(topic))
            this.metadata.add(topic);

        if (metadata.fetch().partitionsForTopic(topic) != null)
            return 0;

        long begin = time.milliseconds();
        long remainingWaitMs = maxWaitMs;
        while (metadata.fetch().partitionsForTopic(topic) == null) {
            log.trace("Requesting metadata update for topic {}.", topic);
            int version = metadata.requestUpdate();
            sender.wakeup();
            metadata.awaitUpdate(version, remainingWaitMs);
            long elapsed = time.milliseconds() - begin;
            if (elapsed >= maxWaitMs)
                throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
            if (metadata.fetch().unauthorizedTopics().contains(topic))
                throw new TopicAuthorizationException(topic);
            remainingWaitMs = maxWaitMs - elapsed;
        }
        return time.milliseconds() - begin;
    }

    /**
     * Get the current cluster info without blocking
     */
    public synchronized Cluster fetch() {
        return this.cluster;
    }

    public synchronized int requestUpdate() {
        this.needUpdate = true;
        return this.version;
    }

    /**
     * Wait for metadata update until the current version is larger than the last version we know of
     */
    public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
        if (maxWaitMs < 0) {
            throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milli seconds");
        }
        long begin = System.currentTimeMillis();
        long remainingWaitMs = maxWaitMs;
        while (this.version <= lastVersion) {
            if (remainingWaitMs != 0)
                wait(remainingWaitMs);
            long elapsed = System.currentTimeMillis() - begin;
            if (elapsed >= maxWaitMs)
                throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
            remainingWaitMs = maxWaitMs - elapsed;
        }
    }

这个方法核心就是判断了是否有Cluster元数据信息，如果没有，进行了如下操作：

1）metadata.requestUpdate(); 更新了一个needUpdate标记，这个值会影响之前maybeUpdate的metadataTimeout的计算，可以让metadataTimeout为0

2）sender.wakeup();唤醒之前nioSelector.select()的阻塞，继续执行

3）metadata.awaitUpdate(version, remainingWaitMs); 主要进行了版本比较，如果不是最新版本，调用了Metadata.wait()方法（wait方法是每个Object都会有的方法，一般和notify或者notifyAll组合使用）

整个过程我直接用图给大家表示一下，如下所示：

整个图就是今天我们分析的关键结果了，这里通过两种阻塞和唤醒机制，一个是NIO中Selector的select()和wakeUp()，一个是MetaData对象的wait()和notifyAll()机制。所以这里要结合之前Sender线程的阻塞逻辑一起来理解。

是不是很有意思一种使用，这里没有用任何线程的join、sleep、wait、park、unpark、notify这些方法。

小结

最后我们简单小结下，这里一节我们主要分析了如下Producer的源码原理：

初始化KafkaProducer时并没有去拉取元数据，但是创建了Selector组件，启动了Sender线程，select阻塞等待请求响应。由于还没有发送任何请求，所以初始化时并没有去真正拉取元数据。

真正拉取元数据是在第一次send方法调用时，会唤醒唤醒Selector之前阻塞的select(),进入第二次while循环，从而发送拉取元数据请求，并且通过Obejct.wait的机制等待60s，等到从Broker拉取元数据成功后，才会继续执行真正的生产消息的请求，否则会报拉取元数据超时异常。

这一节我们只是看到了进行了wait如何等待元数据拉取。

而唤醒Selector的select之后应该会进入第二次while循环

那第二次while循环如何发送请求拉取元数据请求，并且在成功后notifyAll()进行唤醒操作的呢?

让我们下一节继续分析，大家敬请期待！ 我们下一节见！

本文由博客群发一文多发等运营工具平台 OpenWrite 发布