Kafka的原始碼解讀（一）-- 生產者
? 該檔案及之后的的kafka原始碼解讀均以kafka2.4.0版本進行解讀，kafka是用NIO作為通信基礎的，這里不做贅述，如有需要連接NIO基礎的課參考以下鏈接：
https://editor.csdn.net/md/?articleId=113486103
生產者發送資料流程解讀
? 生產者發送訊息的流程簡圖如下：
[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-iEmm4Tn2-1612792370084)(/Users/dengguoqing/IdeaProjects/bigData/筆記/kafka/kafka生產者訊息概述.png)]
發送訊息的簡要流程如下：
- 生產者接受到訊息
- 將訊息封裝為ProducerRecord
- 將訊息序列化
- 獲取訊息的磁區資訊：如果沒有元資料，需要去獲取元資料
- 將訊息放入快取區
- sender讀取快取區的訊息，并進行batch封裝
- 滿足batch發送的訊息，將訊息發送給Broker
發送訊息核心流程剖析圖如下：
[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-u2L3CDrT-1612792370086)(/Users/dengguoqing/IdeaProjects/bigData/筆記/kafka/kafka發送訊息核心流程剖析.png)]
Kafpa原始碼注釋翻譯如下：
用于將訊息發布到Kafka集群的Kafka生產者客戶端，
生產者是執行緒安全的，因此可以多個執行緒共享一個生產者示例，且通常情況下多執行緒共享同一個實體要比每個執行緒一個實體性能要高，
下面是一個包含鍵/值對序列化器的使用案例：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
producer.close();
生產者由一個緩沖空間池(RecordAccumulator)和一個I/O后臺執行緒（Sender）組成，快取空間保存尚未傳輸到服務器的訊息，Sender執行緒負責將這些訊息轉換為請求（batch）并將它們發送到集群， 當生產者close失敗的時候，這些訊息會丟失，
send()方法是異步的， 呼叫時，它將訊息添加到暫掛記錄發送和立即回傳的緩沖區中， 這樣生產者可以將訊息通過批處理來提高效率，
acks引數控制用于確定請求完成的條件， 我們指定的“all”設定是指將訊息同步到所有磁區，這是最慢但最安全的設定，
如果請求失敗，生產者可以自動重試，雖然我們將retries指定為0，但仍然會進行重試， 啟用重試也將打開重復的可能性（有關詳細資訊，請參閱有關郵件傳遞語意 http://kafka.apache.org/documentation.html#semantics的檔案），
生產者為每個磁區維護未發送訊息的緩沖區（一個磁區一個Dequeue）， 這些緩沖區的大小由batch.size配置指定， 增大它可以導致更多的批處理，但是需要更多的記憶體（因為我們通常會為每個活動磁區使用一個緩沖區），
默認情況下，即使緩沖區中還有其他未使用的空間，緩沖區也可以立即發送， 如果想要減少請求數，可以將linger.ms設定為大于0的值，該引數生產者在發送請求之前等待該毫秒數，以希望會有更多記錄來填充，這類似于Nagle在TCP中的演算法， 例如，在上面的代碼段中，由于我們將延遲時間設定為1毫秒，因此很可能所有100條記錄都將在一個請求中發送， 但是，如果我們沒有填充緩沖區，此設定將為我們的請求增加1毫秒的延遲，以等待更多記錄到達， 請注意，時間接近的記錄通常即使linger.ms=0也作為一個批次處理，因此在高負載下將進行批處理，與linger配置無關， 但是，如果不將其設定為大于0的值，則在不處于最大負載的情況下，可能會導致更少的請求和更有效的請求，但以少量的等待時間為代價，
buffer.memory控制生產者緩沖區的記憶體大小，如果訊息的發送速度超過了將訊息發送到服務器的速度，則該緩沖區空間將被耗盡，當緩沖區空間用盡時，其他發送呼叫將阻塞，阻塞時間的閾值由max.block.ms確定，此閾值之后將引發TimeoutException，
key.serializer和value.serializer指示如何將用戶通過其ProducerRecord提供的鍵和值物件轉換為位元組， 您可以將包含的org.apache.kafka.common.serialization.ByteArraySerializer或org.apache.kafka.common.serialization.StringSerializer用于簡單的字串或位元組型別，
從Kafka 0.11開始，KafkaProducer支持兩種附加模式：冪等生產者和事務生產者， 冪等的生成器將Kafka的交付語意從至少一次交付增強到恰好一次交付， 特別是，生產者重試將不再引入重復項， 事務產生器允許應用程式原子地將訊息發送到多個磁區(partition)和主題（topic）
要啟用冪等，必須將enable.idempotence引數設定為true，如果啟用冪等，則retries配置將默認為Integer.MAX_VALUE，而acks配置將默認為all，冪等生產者沒有API更改，因此無需修改現有應用程式即可利用此功能，
為了利用冪等生成器，必須避免重新發送應用程式級別，因為這些應用程式無法重復洗掉， 因此，如果應用程式啟用冪等性，建議將retries配置保留為未設定狀態，因為它將默認為Integer.MAX_VALUE ， 此外，如果send(ProducerRecord)即使無限次重試也回傳錯誤（例如，如果訊息在發送之前已在緩沖區中過期），則建議關閉生產者并檢查最后產生的訊息的內容，以確保不能重復， 最后，生產者只能保證在單個會話中發送的訊息具有冪等性，
要使用事務性生產者和附帶的API，必須設定transactional.id引數屬性，如果設定了transactional.id ，則會自動啟用冪等性以及冪等性所依賴的生產者配置，此外交易中包含的主題應配置為具有持久性，特別是， replication.factor應該至少為3 ，并且這些主題的min.insync.replicas應該設定為2，最后，為了從端到端實作事務保證，必須使使用者配置為僅讀取已提交的訊息，transactional.id的目的是啟用單個生產者實體的多個會話之間的事務恢復， 它通常從磁區的有狀態應用程式中的分片識別符號派生， 這樣，它對于磁區應用程式中運行的每個生產者實體都應該是唯一的，所有新的事務性API都將被阻止，并且將在失敗時引發例外， 下面的示例說明了如何使用新的API， 除了所有100條訊息都是單個事務的一部分外，它與上面的示例相似，
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
producer.initTransactions();
try {
producer.beginTransaction();
for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
// We can't recover from these exceptions, so our only option is to close the producer and exit.
producer.close();
} catch (KafkaException e) {
// For all other exceptions, just abort the transaction and try again.
producer.abortTransaction();
}
producer.close();
如示例所示，每個生產者只能有一個未完成的事務， 在beginTransaction()和commitTransaction()呼叫之間發送的所有訊息都將成為單個事務的一部分， 指定transactional.id ，生產者發送的所有訊息都必須是事務的一部分，
事務性生產者使用例外來傳達錯誤狀態， 特別是，不需要為producer.send()指定回呼或在回傳的Future上呼叫.get() ：如果在producer.send()任何producer.send()或事務呼叫遇到不可恢復的錯誤，都將拋出KafkaException ，交易， 有關從事務性發送中檢測錯誤的更多詳細資訊，請參見send(ProducerRecord)檔案，通過在接收到KafkaException呼叫producer.abortTransaction() ，我們可以確保將任何成功的寫入標記為已中止，從而保留事務保證，
生產者初始化原始碼決議
kafka對外提供了如下四個構造器:
-
引數封裝為Map傳入
-
public KafkaProducer(final Map<String, Object> configs) { this(configs, null, null, null, null, null, Time.SYSTEM); } -
其他引數封裝為map,序列化器單獨傳入
-
public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer) { this(configs, keySerializer, valueSerializer, null, null, null, Time.SYSTEM); } -
引數封裝為property傳入
-
public KafkaProducer(Properties properties) { this(propsToMap(properties), null, null, null, null, null, Time.SYSTEM); } -
序列化器單獨傳入，其他引數封裝為property
-
public KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer) { this(propsToMap(properties), keySerializer, valueSerializer, null, null, null, Time.SYSTEM); }
所有上述構造器，都是通過呼叫如下構造器完成生產者初始化：
KafkaProducer(Map<String, Object> configs,
Serializer<K> keySerializer,
Serializer<V> valueSerializer,
ProducerMetadata metadata,
KafkaClient kafkaClient,
ProducerInterceptors interceptors,
Time time)
下面分析該方法內容：
KafkaProducer(Map<String, Object> configs,
Serializer<K> keySerializer,
Serializer<V> valueSerializer,
ProducerMetadata metadata,
KafkaClient kafkaClient,
ProducerInterceptors interceptors,
Time time) {
/**
* 如果引數中傳入了鍵值的序列化器，這里進行序列化器配置
*/
ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(configs, keySerializer,
valueSerializer));
try {
/**
* 獲取配置引數
*/
Map<String, Object> userProvidedConfigs = config.originals();
this.producerConfig = config;
this.time = time;
//獲取事務ID
String transactionalId = userProvidedConfigs.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG) ?
(String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG) : null;
//獲取clientId
this.clientId = buildClientId(config.getString(ProducerConfig.CLIENT_ID_CONFIG), transactionalId);
//日志配置
LogContext logContext;
if (transactionalId == null)
logContext = new LogContext(String.format("[Producer clientId=%s] ", clientId));
else
logContext = new LogContext(String.format("[Producer clientId=%s, transactionalId=%s] ", clientId, transactionalId));
log = logContext.logger(KafkaProducer.class);
log.trace("Starting the Kafka producer");
/*
測量工具
*/
Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG))
.timeWindow(config.getLong(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
.recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.tags(metricTags);
List<MetricsReporter> reporters = config.getConfiguredInstances(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class,
Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));
reporters.add(new JmxReporter(JMX_PREFIX));
this.metrics = new Metrics(metricConfig, reporters, time);
//磁區器，若沒有配置，使用DefaultPartitioner作為磁區器
this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
//發送失敗后，多久后重試
long retryBackOffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);
//獲取鍵值序列化類
if (keySerializer == null) {
this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
Serializer.class);
this.keySerializer.configure(config.originals(), true);
} else {
config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
this.keySerializer = keySerializer;
}
if (valueSerializer == null) {
this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
Serializer.class);
this.valueSerializer.configure(config.originals(), false);
} else {
config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
this.valueSerializer = valueSerializer;
}
// load interceptors and make sure they get clientId，加載攔截器
userProvidedConfigs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
ProducerConfig configWithClientId = new ProducerConfig(userProvidedConfigs, false);
List<ProducerInterceptor<K, V>> interceptorList = (List) configWithClientId.getConfiguredInstances(
ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class);
if (interceptors != null)
this.interceptors = interceptors;
else
this.interceptors = new ProducerInterceptors<>(interceptorList);
ClusterResourceListeners clusterResourceListeners = configureClusterResourceListeners(keySerializer,
valueSerializer, interceptorList, reporters);
//單個請求的最大大小，默認值的大小是1M.
this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);
//該值是快取大小，默認是32M.一般夠用，根據實際情況看是否需要修改
this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);
//設定壓縮格式
this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
this.maxBlockTimeMs = config.getLong(ProducerConfig.MAX_BLOCK_MS_CONFIG);
/**
* 事務管理器
*/
this.transactionManager = configureTransactionState(config, logContext, log);
int deliveryTimeoutMs = configureDeliveryTimeout(config, log);
this.apiVersions = new ApiVersions();
//快取器初始化
this.accumulator = new RecordAccumulator(logContext,
config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
this.compressionType,
lingerMs(config),
retryBackOffMs,
deliveryTimeoutMs,
metrics,
PRODUCER_METRIC_GROUP_NAME,
time,
apiVersions,
transactionManager,
//記憶體池
new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));
List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(
config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
config.getString(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG));
//初始化原資料資訊，但是并不真正的加載元資料
if (metadata != null) {
this.metadata = metadata;
} else {
this.metadata = new ProducerMetadata(retryBackOffMs,
config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),
logContext,
clusterResourceListeners,
Time.SYSTEM);
this.metadata.bootstrap(addresses, time.milliseconds());
}
this.errors = this.metrics.sensor("errors");
//sender,真正發送資料的執行緒
this.sender = newSender(logContext, kafkaClient, this.metadata);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();
config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
log.debug("Kafka producer started");
} catch (Throwable t) {
// call close methods if internal objects are already constructed this is to prevent resource leak. see KAFKA-2121
close(Duration.ofMillis(0), true);
// now propagate the exception
throw new KafkaException("Failed to construct kafka producer", t);
}
}
? 生產者雖然有加載元資料的代碼，但是這里并沒有真正的建立和kafka集群的聯系，真正加載元資料是在第一個訊息發送時，獲取磁區資訊時加載的，
? 以上代碼中，最重要的部分是，緩沖區的初始化，以及sender執行緒的啟動，
緩沖區初始化
緩沖區是用來存放已經生產但是還沒有發送給kafka集群的資訊，緩沖區初始化代碼如下：
this.accumulator = new RecordAccumulator(logContext,
config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
this.compressionType,
lingerMs(config),
retryBackOffMs,
deliveryTimeoutMs,
metrics,
PRODUCER_METRIC_GROUP_NAME,
time,
apiVersions,
transactionManager,
//記憶體池
new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time,PRODUCER_METRIC_GROUP_NAME));
sender初始化
? sender執行緒是完成資料從生產者發送到kafka集群的執行緒，該執行緒會從上面的緩沖區拉取可以發送的訊息，并按partition組成發送的batch，之后從生產者發送到kafka集群，代碼如下：
//sender,真正發送資料的執行緒
this.sender = newSender(logContext, kafkaClient, this.metadata);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();
sender初始化代碼如下：
// visible for testing
Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
int maxInflightRequests = configureInflightRequests(producerConfig, transactionManager != null);
int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);
ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(producerConfig, time);
ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);
//網路連接 初始化kafkaClient
KafkaClient client = kafkaClient != null ? kafkaClient : new NetworkClient(
new Selector(producerConfig.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
this.metrics, time, "producer", channelBuilder, logContext),
metadata,
clientId,
maxInflightRequests,
producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),
producerConfig.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
producerConfig.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),
requestTimeoutMs,
ClientDnsLookup.forConfig(producerConfig.getString(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG)),
time,
true,
apiVersions,
throttleTimeSensor,
logContext);
int retries = configureRetries(producerConfig, transactionManager != null, log);
short acks = configureAcks(producerConfig, transactionManager != null, log);
/**
* acks:
* 0: 生產者發送完資料，不管Broker的回應
* 1: 生產者發送資料后，接受到Leader Partition回傳后成功
* -1: 生產者發送資料后，接收到所有partition成功后，才認為成功
*/
return new Sender(logContext,
client,
metadata,
this.accumulator,
maxInflightRequests == 1,
producerConfig.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
acks,
retries,
metricsRegistry.senderMetrics,
time,
requestTimeoutMs,
producerConfig.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
this.transactionManager,
apiVersions);
}
生產者發送訊息代碼決議
生產者發送訊息的方法有兩個分別如下:
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
return send(record, null);
}
@Override
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.onSend(record);
return doSend(interceptedRecord, callback);
}
? 可以看出最后都是呼叫第二個方法，該方法會將訊息異步發送到指定topic并在訊息確認發送成功后呼叫提供的回呼函式，該方法在訊息放到生產者的緩沖區后就立刻回傳，這樣可以并行無阻塞的發送訊息，
? 該方法的回傳值是一個RecordMetadata,該類包含訊息的磁區資訊、訊息的offset、以及創建訊息的時間，topic指定的TimestampType是CREATE_TIME時，訊息的timestamp是用戶提供的一個時間戳，如果用戶沒有指定時間戳，則使用訊息發送的時間，如果topic指定的TimestampType是LOG_APPEND_TIME是LogAppendTime，訊息的時間戳是訊息添加到Kafka集群時的本地時間，
? 上述方法最侄訓進入doSend()方法，下面各處doSend()方法的決議，
//檢查生產者是否已經關閉
throwIfProducerClosed();
// first make sure the metadata for the topic is available
//確定已經獲取到元資料，且topic是可用的，根據指定的topic和磁區資訊獲取元資料
ClusterAndWaitTime clusterAndWaitTime;
/**
* 步驟一：獲取元資料，maxBlockTimeMs最大等待時間
*/
try {
clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
} catch (KafkaException e) {
if (metadata.isClosed())
throw new KafkaException("Producer closed while send in progress", e);
throw e;
}
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
Cluster cluster = clusterAndWaitTime.cluster;
/**
* 步驟二：
* 對訊息的key和value進行序列化
*/
byte[] serializedKey;
try {
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 {
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);
}
/**
* 步驟三：獲取磁區
*/
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();
if (log.isTraceEnabled()) {
log.trace("Attempting to append 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.failIfNotReadyForSend();
}
/**
* 步驟六：
* 將訊息追加到緩沖區中，這是一個要終點關注的方法，里面有很多提升性能巧妙的設計
*/
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
serializedValue, headers, interceptCallback, remainingWaitMs, true);
/**
* 當batch已經滿了，要在這里新開辟batch添加資料
*/
if (result.abortForNewBatch) {
int prevPartition = partition;
partitioner.onNewBatch(record.topic(), cluster, prevPartition);
partition = partition(record, serializedKey, serializedValue, cluster);
tp = new TopicPartition(record.topic(), partition);
if (log.isTraceEnabled()) {
log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", record.topic(), partition, prevPartition);
}
// producer callback will make sure to call both 'callback' and interceptor callback
interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);
result = accumulator.append(tp, timestamp, serializedKey,
serializedValue, headers, interceptCallback, remainingWaitMs, false);
}
if (transactionManager != null && transactionManager.isTransactional())
transactionManager.maybeAddPartitionToTransaction(tp);
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;
根據上述代碼可以看出，發送資料的流程大概分為八個步驟，下面對每個步驟進行分析
步驟一：獲取元資料
獲取元資料的代碼
clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long maxWaitMs) throws InterruptedException {
// add topic to metadata topic list if it is not there already and reset expiry
Cluster cluster = metadata.fetch();
//檢查topic是否失效
if (cluster.invalidTopics().contains(topic))
throw new InvalidTopicException(topic);
//將topic添加到元資料
metadata.add(topic);
/**
* 獲取topic的磁區資訊，第一次發送資料的時候，該值為null
*/
Integer partitionsCount = cluster.partitionCountForTopic(topic);
// Return cached metadata if we have it, and if the record's partition is either undefined
// or within the known partition range
if (partitionsCount != null && (partition == null || partition < partitionsCount))
return new ClusterAndWaitTime(cluster, 0);
/**
* 獲取元資料的開始時間
*/
long begin = time.milliseconds();
/**
* 獲取元資料的剩余時間
*/
long remainingWaitMs = maxWaitMs;
/**
* 獲取元資料花費的時間
*/
long elapsed;
// Issue metadata requests until we have metadata for the topic and the requested partition,
// or until maxWaitTimeMs is exceeded. This is necessary in case the metadata
// is stale and the number of partitions for this topic has increased in the meantime.
do {
if (partition != null) {
log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
} else {
log.trace("Requesting metadata update for topic {}.", topic);
}
metadata.add(topic);
int version = metadata.requestUpdate();
/**
* 喚醒sender執行緒，這里是獲取元資料真正位置
*/
sender.wakeup();
try {
/**
* 掛起該執行緒，直到sender執行緒獲取元資料成功，或者超時
*/
metadata.awaitUpdate(version, remainingWaitMs);
} catch (TimeoutException ex) {
// Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
throw new TimeoutException(
String.format("Topic %s not present in metadata after %d ms.",
topic, maxWaitMs));
}
/**
* 以下都是獲取的元資料賦值及檢查
*/
cluster = metadata.fetch();
elapsed = time.milliseconds() - begin;
if (elapsed >= maxWaitMs) {
throw new TimeoutException(partitionsCount == null ?
String.format("Topic %s not present in metadata after %d ms.",
topic, maxWaitMs) :
String.format("Partition %d of topic %s with partition count %d is not present in metadata after %d ms.",
partition, topic, partitionsCount, maxWaitMs));
}
metadata.maybeThrowExceptionForTopic(topic);
remainingWaitMs = maxWaitMs - elapsed;
partitionsCount = cluster.partitionCountForTopic(topic);
} while (partitionsCount == null || (partition != null && partition >= partitionsCount));
return new ClusterAndWaitTime(cluster, elapsed);
}
? 根據這段代碼可以看出，獲取元資料需要sender執行緒獲取，下面分析下sender執行緒獲取元資料的代碼，run()方法中關鍵代碼如下
log.debug("Starting Kafka producer I/O thread.");
/**
* 該方法在Producer關閉前會一直呼叫
*/
while (running) {
try {
runOnce();
} catch (Exception e) {
log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}
run()方法中關鍵的執行方法是runOnce()，關鍵代碼如下：
/**
* 發送資料的代碼在下面，
*/
long currentTimeMs = time.milliseconds();
long pollTimeout = sendProducerData(currentTimeMs);
client.poll(pollTimeout, currentTimeMs);
其中關鍵代碼是poll()方法，關鍵獲取請求的代碼如下：
// process completed actions
long updatedNow = this.time.milliseconds();
List<ClientResponse> responses = new ArrayList<>();
handleCompletedSends(responses, updatedNow);
//處理回傳的請求
handleCompletedReceives(responses, updatedNow);
handleDisconnections(responses, updatedNow);
handleConnections();
handleInitiateApiVersionRequests(updatedNow);
handleTimedOutRequests(responses, updatedNow);
completeResponses(responses);
步驟二：對訊息的key和value進行序列化
這段代碼沒有特殊的邏輯，不再贅述
步驟三：獲取資料磁區
int partition = partition(record, serializedKey, serializedValue, cluster)
Integer partition = record.partition();
return partition != null ?
partition :
partitioner.partition(
record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
if (keyBytes == null) {
return stickyPartitionCache.partition(topic, cluster);
}
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
步驟四：檢查訊息的大小
? 檢查訊息的大小，單條訊息的大小既不能大于設定的最大值，也不能大于緩沖區的大小
if (size > this.maxRequestSize)
throw new RecordTooLargeException("The message is " + size +
" bytes when serialized which is larger than the maximum request size you have configured with the " +
ProducerConfig.MAX_REQUEST_SIZE_CONFIG +
" configuration.");
if (size > this.totalMemorySize)
throw new RecordTooLargeException("The message is " + size +
" bytes when serialized which is larger than the total memory buffer you have configured with the " +
ProducerConfig.BUFFER_MEMORY_CONFIG +
" configuration.");
步驟五：系結回呼函式
Callback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);
步驟六：訊息追加到緩沖區
? 將訊息追加到緩沖區，這里會把訊息封裝為一個batches，這里有很多巧妙的設計，
appendsInProgress.incrementAndGet();
ByteBuffer buffer = null;
if (headers == null) headers = Record.EMPTY_HEADERS;
try {
// 這里檢查對應Topic的Deque是否已經創建
Deque<ProducerBatch> dq = getOrCreateDeque(tp);
// 這里是多執行緒可以同時操作，所以在這里加鎖
synchronized (dq) {
if (closed)
throw new KafkaException("Producer closed while send in progress");
/**
* 將訊息添加到Toic對應的Batch中
*/
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
if (abortOnNewBatch) {
// Return a result that will cause another call to append.
return new RecordAppendResult(null, false, false, true);
}
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 KafkaException("Producer closed while send in progress");
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 = Objects.requireNonNull(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, false);
}
} finally {
if (buffer != null)
free.deallocate(buffer);
appendsInProgress.decrementAndGet();
}
? 代碼中使用了分段加鎖，用來提高性能，其中創建Batches使用了自己創建的資料結構，是一個讀寫分離的Map,也是為了提高性能進行設計的，
? 上述中有記憶體池的設計，記憶體池設計示意圖如下：
[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-dzOIjhR8-1612792370088)(/Users/dengguoqing/IdeaProjects/bigData/筆記/kafka/記憶體池設計.png)]
步驟七：當滿足發送資料的條件時，進行資料的發送
即batches滿足大小，或者時間達到了發送的條件，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/258153.html
標籤：其他