1、背景引入：很多同學(xué)看不懂kafka參數(shù)

今天給大家聊一個(gè)很有意思的話題，大家知道很多公司都會(huì)基于Kafka作為MQ來(lái)開發(fā)一些復(fù)雜的大型系統(tǒng)。

而在使用Kafka的客戶端編寫代碼與服務(wù)器交互的時(shí)候，是需要對(duì)客戶端設(shè)置很多的參數(shù)的。

所以我就見過(guò)很多年輕的同學(xué)，可能剛剛加入團(tuán)隊(duì)，對(duì)Kafka這個(gè)技術(shù)其實(shí)并不是很了解。

此時(shí)就會(huì)導(dǎo)致他們看團(tuán)隊(duì)里的一些資深同事寫的一些代碼，會(huì)看不懂是怎么回事，不了解背后的含義，這里面尤其是一些Kafka參數(shù)的設(shè)置。

所以這篇文章，我們還是采用老規(guī)矩畫圖的形式，來(lái)聊聊Kafka生產(chǎn)端一些常見參數(shù)的設(shè)置，讓大家下次看到一些Kafka客戶端設(shè)置的參數(shù)時(shí)，不會(huì)再感到發(fā)怵。

2、一段Kafka生產(chǎn)端的示例代碼

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("buffer.memory", 67108864);
props.put("batch.size", 131072);
props.put("linger.ms", 100);
props.put("max.request.size", 10485760);
props.put("acks", "1");
props.put("retries", 10);
props.put("retry.backoff.ms", 500);

KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

3、內(nèi)存緩沖的大小

首先我們看看“buffer.memory”這個(gè)參數(shù)是什么意思？

Kafka的客戶端發(fā)送數(shù)據(jù)到服務(wù)器，一般都是要經(jīng)過(guò)緩沖的，也就是說(shuō)，你通過(guò)KafkaProducer發(fā)送出去的消息都是先進(jìn)入到客戶端本地的內(nèi)存緩沖里，然后把很多消息收集成一個(gè)一個(gè)的Batch，再發(fā)送到Broker上去的。

所以這個(gè)“buffer.memory”的本質(zhì)就是用來(lái)約束KafkaProducer能夠使用的內(nèi)存緩沖的大小的，他的默認(rèn)值是32MB。

那么既然了解了這個(gè)含義，大家想一下，在生產(chǎn)項(xiàng)目里，這個(gè)參數(shù)應(yīng)該怎么來(lái)設(shè)置呢？

你可以先想一下，如果這個(gè)內(nèi)存緩沖設(shè)置的過(guò)小的話，可能會(huì)導(dǎo)致一個(gè)什么問(wèn)題？

首先要明確一點(diǎn)，那就是在內(nèi)存緩沖里大量的消息會(huì)緩沖在里面，形成一個(gè)一個(gè)的Batch，每個(gè)Batch里包含多條消息。

然后KafkaProducer有一個(gè)Sender線程會(huì)把多個(gè)Batch打包成一個(gè)Request發(fā)送到Kafka服務(wù)器上去。

?那么如果要是內(nèi)存設(shè)置的太小，可能導(dǎo)致一個(gè)問(wèn)題：消息快速的寫入內(nèi)存緩沖里面，但是Sender線程來(lái)不及把Request發(fā)送到Kafka服務(wù)器。

這樣是不是會(huì)造成內(nèi)存緩沖很快就被寫滿？一旦被寫滿，就會(huì)阻塞用戶線程，不讓繼續(xù)往Kafka寫消息了。

所以對(duì)于“buffer.memory”這個(gè)參數(shù)應(yīng)該結(jié)合自己的實(shí)際情況來(lái)進(jìn)行壓測(cè)，你需要測(cè)算一下在生產(chǎn)環(huán)境，你的用戶線程會(huì)以每秒多少消息的頻率來(lái)寫入內(nèi)存緩沖。?

比如說(shuō)每秒300條消息，那么你就需要壓測(cè)一下，假設(shè)內(nèi)存緩沖就32MB，每秒寫300條消息到內(nèi)存緩沖，是否會(huì)經(jīng)常把內(nèi)存緩沖寫滿？經(jīng)過(guò)這樣的壓測(cè)，你可以調(diào)試出來(lái)一個(gè)合理的內(nèi)存大小。

4、多少數(shù)據(jù)打包為一個(gè)Batch合適？

接著你需要思考第二個(gè)問(wèn)題，就是你的“batch.size”應(yīng)該如何設(shè)置？這個(gè)東西是決定了你的每個(gè)Batch要存放多少數(shù)據(jù)就可以發(fā)送出去了。

比如說(shuō)你要是給一個(gè)Batch設(shè)置成是16KB的大小，那么里面湊夠16KB的數(shù)據(jù)就可以發(fā)送了。

這個(gè)參數(shù)的默認(rèn)值是16KB，一般可以嘗試把這個(gè)參數(shù)調(diào)節(jié)大一些，然后利用自己的生產(chǎn)環(huán)境發(fā)消息的負(fù)載來(lái)測(cè)試一下。

比如說(shuō)發(fā)送消息的頻率就是每秒300條，那么如果比如“batch.size”調(diào)節(jié)到了32KB，或者64KB，是否可以提升發(fā)送消息的整體吞吐量。

因?yàn)槔碚撋蟻?lái)說(shuō)，提升batch的大小，可以允許更多的數(shù)據(jù)緩沖在里面，那么一次Request發(fā)送出去的數(shù)據(jù)量就更多了，這樣吞吐量可能會(huì)有所提升。

但是這個(gè)東西也不能無(wú)限的大，過(guò)于大了之后，要是數(shù)據(jù)老是緩沖在Batch里遲遲不發(fā)送出去，那么豈不是你發(fā)送消息的延遲就會(huì)很高。

比如說(shuō)，一條消息進(jìn)入了Batch，但是要等待5秒鐘Batch才湊滿了64KB，才能發(fā)送出去。那這條消息的延遲就是5秒鐘。

所以需要在這里按照生產(chǎn)環(huán)境的發(fā)消息的速率，調(diào)節(jié)不同的Batch大小自己測(cè)試一下最終出去的吞吐量以及消息的 延遲，設(shè)置一個(gè)最合理的參數(shù)。

5、要是一個(gè)Batch遲遲無(wú)法湊滿怎么辦？

要是一個(gè)Batch遲遲無(wú)法湊滿，此時(shí)就需要引入另外一個(gè)參數(shù)了，“l(fā)inger.ms”

他的含義就是說(shuō)一個(gè)Batch被創(chuàng)建之后，最多過(guò)多久，不管這個(gè)Batch有沒(méi)有寫滿，都必須發(fā)送出去了。

給大家舉個(gè)例子，比如說(shuō)batch.size是16kb，但是現(xiàn)在某個(gè)低峰時(shí)間段，發(fā)送消息很慢。

這就導(dǎo)致可能Batch被創(chuàng)建之后，陸陸續(xù)續(xù)有消息進(jìn)來(lái)，但是遲遲無(wú)法湊夠16KB，難道此時(shí)就一直等著嗎？

當(dāng)然不是，假設(shè)你現(xiàn)在設(shè)置“l(fā)inger.ms”是50ms，那么只要這個(gè)Batch從創(chuàng)建開始到現(xiàn)在已經(jīng)過(guò)了50ms了，哪怕他還沒(méi)滿16KB，也要發(fā)送他出去了。

所以“l(fā)inger.ms”決定了你的消息一旦寫入一個(gè)Batch，最多等待這么多時(shí)間，他一定會(huì)跟著Batch一起發(fā)送出去。

避免一個(gè)Batch遲遲湊不滿，導(dǎo)致消息一直積壓在內(nèi)存里發(fā)送不出去的情況。這是一個(gè)很關(guān)鍵的參數(shù)。

這個(gè)參數(shù)一般要非常慎重的來(lái)設(shè)置，要配合batch.size一起來(lái)設(shè)置。

舉個(gè)例子，首先假設(shè)你的Batch是32KB，那么你得估算一下，正常情況下，一般多久會(huì)湊夠一個(gè)Batch，比如正常來(lái)說(shuō)可能20ms就會(huì)湊夠一個(gè)Batch。

那么你的linger.ms就可以設(shè)置為25ms，也就是說(shuō)，正常來(lái)說(shuō)，大部分的Batch在20ms內(nèi)都會(huì)湊滿，但是你的linger.ms可以保證，哪怕遇到低峰時(shí)期，20ms湊不滿一個(gè)Batch，還是會(huì)在25ms之后強(qiáng)制Batch發(fā)送出去。

如果要是你把linger.ms設(shè)置的太小了，比如說(shuō)默認(rèn)就是0ms，或者你設(shè)置個(gè)5ms，那可能導(dǎo)致你的Batch雖然設(shè)置了32KB，但是經(jīng)常是還沒(méi)湊夠32KB的數(shù)據(jù)，5ms之后就直接強(qiáng)制Batch發(fā)送出去，這樣也不太好其實(shí)，會(huì)導(dǎo)致你的Batch形同虛設(shè)，一直湊不滿數(shù)據(jù)。

6、最大請(qǐng)求大小

“max.request.size”這個(gè)參數(shù)決定了每次發(fā)送給Kafka服務(wù)器請(qǐng)求的最大大小，同時(shí)也會(huì)限制你一條消息的最大大小也不能超過(guò)這個(gè)參數(shù)設(shè)置的值，這個(gè)其實(shí)可以根據(jù)你自己的消息的大小來(lái)靈活的調(diào)整。

給大家舉個(gè)例子，你們公司發(fā)送的消息都是那種大的報(bào)文消息，每條消息都是很多的數(shù)據(jù)，一條消息可能都要20KB。

此時(shí)你的batch.size是不是就需要調(diào)節(jié)大一些？比如設(shè)置個(gè)512KB？然后你的buffer.memory是不是要給的大一些？比如設(shè)置個(gè)128MB？

只有這樣，才能讓你在大消息的場(chǎng)景下，還能使用Batch打包多條消息的機(jī)制。但是此時(shí)“max.request.size”是不是也得同步增加？

因?yàn)榭赡苣愕囊粋€(gè)請(qǐng)求是很大的，默認(rèn)他是1MB，你是不是可以適當(dāng)調(diào)大一些，比如調(diào)節(jié)到5MB？

7、重試機(jī)制

“retries”和“retries.backoff.ms”決定了重試機(jī)制，也就是如果一個(gè)請(qǐng)求失敗了可以重試幾次，每次重試的間隔是多少毫秒。

這個(gè)大家適當(dāng)設(shè)置幾次重試的機(jī)會(huì)，給一定的重試間隔即可，比如給100ms的重試間隔。

8、持久化機(jī)制

“acks”參數(shù)決定了發(fā)送出去的消息要采用什么樣的持久化策略，這個(gè)涉及到了很多其他的概念。