關於高併發下kafka producer send非同步傳送耗時問題的分析

摘要： 最近開發閘道器服務的過程當中，需要用到kafka轉發訊息與儲存日誌，在進行壓測的過程中由於是多執行緒併發操作kafka producer 進行非同步send，發現send耗時有時會達到幾十毫秒的阻塞，很大程度上上影響了併發的效能，而在後續的測試中發現單執行緒傳送反而比多執行緒傳送效率高出幾倍。...

最近開發閘道器服務的過程當中，需要用到kafka轉發訊息與儲存日誌，在進行壓測的過程中由於是多執行緒併發操作kafka producer 進行非同步send，發現send耗時有時會達到幾十毫秒的阻塞，很大程度上上影響了併發的效能，而在後續的測試中發現單執行緒傳送反而比多執行緒傳送效率高出幾倍。所以就對kafka API send 的原始碼進行了一下跟蹤和分析，在此總結記錄一下。

首先看springboot下 kafka producer 的使用

在config中進行配置，向IOC容器中注入DefaultKafkaProducerFactory生產者工廠的例項

@Bean
public ProducerFactory<Object, Object> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}

建立producer

this.producer = producerFactory.createProducer();

大家都知道springboot下IOC容器管理的例項預設都是單例模式；而DefaultKafkaProducerFactory本身也是一個單例工廠

@Override
public Producer<K, V> createProducer() {
if (this.transactionIdPrefix != null) {
return createTransactionalProducer();
}
if (this.producer == null) {
synchronized (this) {
if (this.producer == null) {
this.producer = new CloseSafeProducer<K, V>(createKafkaProducer());
}
}
}
return this.producer;
}

我們建立的producer也是個單例。

接下來就是具體的傳送，用過kafka的小夥伴都知道producer.send是個非同步操作，會返回一個Future<RecordMetadata> 型別的結果。那麼為什麼單執行緒和多執行緒send效率會較大的差距呢，我們進入KafkaProducer內部看下producer.send的具體原始碼實現來找下答案

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
//保證主題的元資料可用
ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
Cluster cluster = clusterAndWaitTime.cluster;
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);
}
//計算出具體的partition 
int partition = partition(record, serializedKey, serializedValue, cluster);
tp = new TopicPartition(record.topic(), partition);

setReadOnly(record.headers());
Header[] headers = record.headers().toArray();

int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),
compressionType, serializedKey, serializedValue, headers);
ensureValidRecordSize(serializedSize);
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 = new InterceptorCallback<>(callback, this.interceptors, tp);

if (transactionManager != null && transactionManager.isTransactional())
transactionManager.maybeAddPartitionToTransaction(tp);
//向佇列容器中新增資料
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
serializedValue, headers, 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();
this.interceptors.onSendError(record, tp, e);
return new FutureFailure(e);
} catch (InterruptedException e) {
this.errors.record();
this.interceptors.onSendError(record, tp, e);
throw new InterruptException(e);
} catch (BufferExhaustedException e) {
this.errors.record();
this.metrics.sensor("buffer-exhausted-records").record();
this.interceptors.onSendError(record, tp, e);
throw e;
} catch (KafkaException e) {
this.errors.record();
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
this.interceptors.onSendError(record, tp, e);
throw e;
}
}

這裡除了前面做的一些序列化操作和判斷，最關鍵的就是向佇列容器中執行新增資料操作

RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
serializedValue, headers, interceptCallback, remainingWaitMs);

accumulator是RecordAccumulator這個類的一個例項，RecordAccumulator類是一個佇列容器類； 它的內部維護了一個 ConcurrentMap，每一個 TopicPartition都對應一個專屬的訊息佇列。

private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

我們進入accumulator.append內部看下具體的實現

public RecordAppendResult append(TopicPartition tp,
long timestamp,
byte[] key,
byte[] value,
Header[] headers,
Callback callback,
long maxTimeToBlock) throws InterruptedException {
// We keep track of the number of appending thread to make sure we do not miss batches in
// abortIncompleteBatches().
appendsInProgress.incrementAndGet();
ByteBuffer buffer = null;
if (headers == null) headers = Record.EMPTY_HEADERS;
try {
//根據TopicPartition拿到對應的批處理佇列 
Deque<ProducerBatch> dq = getOrCreateDeque(tp);
//同步佇列，保證執行緒安全
synchronized (dq) {
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
//把序列化後的資料放入佇列，並返回結果
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
if (appendResult != null)
return appendResult;
}

// we don't have an in-progress record batch try to allocate a new batch
byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
log.trace("Allocating a new {} byte message buffer for topic {} partition {}", size, tp.topic(), tp.partition());
buffer = free.allocate(size, maxTimeToBlock);
synchronized (dq) {
// Need to check if producer is closed again after grabbing the dequeue lock.
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");

RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
if (appendResult != null) {
// Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
return appendResult;
}

MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic);
ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds());
FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()));

dq.addLast(batch);
incomplete.add(batch);

// Don't deallocate this buffer in the finally block as it's being used in the record batch
buffer = null;

return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true);
}
} finally {
if (buffer != null)
free.deallocate(buffer);
appendsInProgress.decrementAndGet();
}
}

在getOrCreateDeque中我們根據TopicPartition從ConcurrentMap獲取對應佇列，沒有的話就初始化一個。

private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
Deque<ProducerBatch> d = this.batches.get(tp);
if (d != null)
return d;
d = new ArrayDeque<>();
Deque<ProducerBatch> previous = this.batches.putIfAbsent(tp, d);
if (previous == null)
return d;
else
return previous;
}

更關鍵的是為了保證併發時的執行緒安全，執行 RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq)時， Deque <ProducerBatch>必然需要同步處理。 

synchronized (dq) {
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
if (appendResult != null)
return appendResult;
}

在這裡我們可以看出，多執行緒高併發情況下，dq會處在比較大的資源競爭，雖然是基於記憶體的操作，每個執行緒持有鎖的時間極短，但相比單執行緒情況，高併發情況下執行緒開闢較多，鎖競爭和cpu上下文切換都比較頻繁，會造成一定的效能損耗，產生阻塞耗時。

分析到這裡你就會發現，其實KafkaProducer這個非同步傳送是建立在生產者和消費者模式上的，send的真正操作並不是直接非同步傳送，而是把資料放在一箇中間佇列中。那麼既然有生產者在往記憶體佇列中放入資料，那麼必然會有一個專有的執行緒負責把這些資料真正傳送出去。我們通過監控jvm執行緒資訊可以看到，KafkaProducer建立後確實會啟動一個守護執行緒用於訊息的傳送。

OK，我們再回到 KafkaProducer中，會看到裡面有這樣兩個物件，Sender就是kafka傳送資料的後臺執行緒

private final Sender sender;
private final Thread ioThread;

在KafkaProducer的建構函式中會啟動Sender執行緒

this.sender = new Sender(logContext,
client,
this.metadata,
this.accumulator,
maxInflightRequests == 1,
config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
acks,
retries,
metricsRegistry.senderMetrics,
Time.SYSTEM,
this.requestTimeoutMs,
config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
this.transactionManager,
apiVersions);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

進入Sender內部可以看到這個執行緒的作用就是一直輪詢傳送資料。

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.hasUndrained() || 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.
log.debug("Aborting incomplete batches due to forced shutdown");
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 a single iteration of sending
*
* @param now The current POSIX time in milliseconds
*/
void run(long now) {
if (transactionManager != null) {
try {
if (transactionManager.shouldResetProducerStateAfterResolvingSequences())
// Check if the previous run expired batches which requires a reset of the producer state.
transactionManager.resetProducerId();

if (!transactionManager.isTransactional()) {
// this is an idempotent producer, so make sure we have a producer id
maybeWaitForProducerId();
} else if (transactionManager.hasUnresolvedSequences() && !transactionManager.hasFatalError()) {
transactionManager.transitionToFatalError(new KafkaException("The client hasn't received acknowledgment for " +
"some previously sent messages and can no longer retry them. It isn't safe to continue."));
} else if (transactionManager.hasInFlightTransactionalRequest() || maybeSendTransactionalRequest(now)) {
// as long as there are outstanding transactional requests, we simply wait for them to return
client.poll(retryBackoffMs, now);
return;
}

// do not continue sending if the transaction manager is in a failed state or if there
// is no producer id (for the idempotent case).
if (transactionManager.hasFatalError() || !transactionManager.hasProducerId()) {
RuntimeException lastError = transactionManager.lastError();
if (lastError != null)
maybeAbortBatches(lastError);
client.poll(retryBackoffMs, now);
return;
} else if (transactionManager.hasAbortableError()) {
accumulator.abortUndrainedBatches(transactionManager.lastError());
}
} catch (AuthenticationException e) {
// This is already logged as error, but propagated here to perform any clean ups.
log.trace("Authentication exception while processing transactional request: {}", e);
transactionManager.authenticationFailed(e);
}
}

long pollTimeout = sendProducerData(now);
client.poll(pollTimeout, now);
}

通過上面的分析我們可以看出producer.send操作本身其實是個基於記憶體的儲存操作，耗時幾乎可以忽略不計，但由於高併發情況下，執行緒同步會有一定的效能損耗，當然這個損耗在一般的應用場景下幾乎是可以忽略不計的，但如果是資料量比較大，高併發的場景下會比較明顯。

針對上面的問題分析，這裡說下我個人的一些總結：

1、首先避免多執行緒操作producer傳送資料，你可以採用生產者消費者模式把producer.send從你的多執行緒操作中解耦出來，維護一個你要傳送的訊息佇列，單獨開闢一個執行緒操作；

2、可能有的小夥伴會問，那麼多建立幾個producer的例項或者維護一個producer池可以嗎，我原本也是這個想法，只是在測試中發現效果也不是很理想，我估計是由於建立producer例項過多，導致執行緒數量也跟著增加，本身的業務執行緒再加上kafka的執行緒，執行緒上下文切換比較頻繁，CPU資源壓力比較大，效率也不如單執行緒操作；

3、這個問題其實真是針對API操作來講的，send操作並不是真正的資料傳送，真正的資料傳送由守護執行緒進行；按照kafka本身的設計思想，如果操作本身就成為了你效能的瓶頸，你應該考慮的是叢集部署，負載均衡；

4、無鎖才是真正的高效能；