Shulou(Shulou.com)06/01 Report--

This article shows you how to carry out kafka asynchronous mode analysis, the content is concise and easy to understand, can definitely brighten your eyes, through the detailed introduction of this article, I hope you can get something.

What is asynchronous mode?

The kafka producer can choose to send data asynchronously, which means we call the send () method and specify a callback function that the server calls when it returns a response.

Kafka exposes two send methods on the client side, and we can choose either synchronous or asynchronous mode. Let's take a look at an example sent by a producer of kafka, which has an intuitive feeling. This example is a synchronous mode.

ProducerRecord record = new ProducerRecord ("Kafka", "Kafka_Products", "test"); / / Topic Key Value

Try {

Future future = producer.send (record)

Future.get (); / / get the execution result

} catch (Exception e) {

E.printStackTrace ()

}

Let's continue to look at it from the source level.

First, kafka defines an interface

Then KafkaProducer implements these two methods, and let's take a look at the implementation logic of the asynchronous method.

You can see that the doSend method is finally called, passing in a callback. This callback is the result of listening for the execution of the method.

Asynchronous mode will also block.

Many people will think that since it is asynchronous mode, no matter whether the result is success or failure, the method call will return immediately. Then I can only tell you, I'm sorry, that's not necessarily the case. I've stepped on this hole myself.

At that time, we had a business process that needed to send kakfa messages to some business party after the execution was completed. In order to minimize the execution time of my main process, we sent kafka messages asynchronously. In use, because the TOPIC information of kafka is mismatched, it is found that the process blocks sending messages for as long as 6 seconds (the default send timeout for kafka).

Why on earth does asynchronous mode still block? Let's keep looking at the source code.

Whether it's synchronous mode or asynchronous mode, the doSend method will eventually be called. Notice the waitOnMetadata method in the figure above. The blocking situation I mentioned above is blocking in this method. Then let's continue to look at this method.

We can probably understand the function of this method through the comments in the code, but I still want to explain it here. (in case someone can't read English, )

WaitOnMetadata gets the current cluster metadata information. If the cache exists and the partition does not exceed the specified partition range, the cache returns, otherwise it triggers an update and waits for a new metadata. The waiting operation is in the following line of code:

Metadata.awaitUpdate (version, remainingWaitMs)

And then continue to follow.

This method is easy to understand, that is, you have been waiting for a condition that will be returned when the condition is met, otherwise you will be waiting for a timeout to exit. The condition is that the current version number is greater than the previous version number.

So who will update the version number? This is the sender thread we mentioned earlier. When our topic was misconfigured, the metadata could not be updated, and then waited until the timeout.

The above is how to analyze the asynchronous mode of kafka. Have you learned any knowledge or skills? If you want to learn more skills or enrich your knowledge reserve, you are welcome to follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.