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 发布！