一、github下載kafka的源碼

可以看到kafka的源碼開源社區(qū)是非常活躍的。

二、搭建kafka環(huán)境

構(gòu)建kafka環(huán)境，首先需要安裝Scala和gradle，再安裝的scala插件需要和你的idea需要匹配，同時不要忘了安裝gradle時配置環(huán)境變量。

安裝完之后，你就可以修改gradle.properties

group=org.apache.kafka
#?NOTE:?When?you?change?this?version?number,?you?should?also?make?sure?to?update
#?the?version?numbers?in
#??-?docs/js/templateData.js
#??-?tests/kafkatest/__init__.py
#??-?tests/kafkatest/version.py?(variable?DEV_VERSION)
#??-?kafka-merge-pr.py
version=1.1.2-SNAPSHOT
scalaVersion=2.11.12
task=build
org.gradle.jvmargs=-Xmx2g?-Xss4m?-XX:+UseParallelGC

修改完之后，就可以進行構(gòu)建了。此時你需要輸入gradle idea進行編譯，這里由于我編譯過了，所以時間較短，通常會較長時間。

重點關(guān)注example包：

三、生產(chǎn)者

生產(chǎn)者producer：

可以看到生產(chǎn)者里面有生產(chǎn)者、主題、是否是異步的相關(guān)變量，同時三個變量都是final，從而我們可以確切的直到它們都是在創(chuàng)建的時候就需要進行指定。構(gòu)造函數(shù)中首先填充配置信息，將配置信息放入創(chuàng)建的kafka生產(chǎn)者中，將主題賦值給topic、生產(chǎn)者是否異步發(fā)送消息放入到isAsync。

重點在run方法中的send方法，發(fā)送方式分為同步和異步兩種方式。

/**
?*?生產(chǎn)者
?*/
public?class?Producer?extends?Thread?{
????//生產(chǎn)者
????private?final?KafkaProducer?producer;//主題private?final?String?topic;//是否是異步private?final?Boolean?isAsync;//構(gòu)造函數(shù)：配置信息服務(wù)器配置、客戶端id、key序列化、value序列化、創(chuàng)建kafka生產(chǎn)者對象、主題、是否是異步public?Producer(String?topic,?Boolean?isAsync)?{
????????Properties?props?=?new?Properties();
????????props.put("bootstrap.servers",?KafkaProperties.KAFKA_SERVER_URL?+?":"?+?KafkaProperties.KAFKA_SERVER_PORT);
????????props.put("client.id",?"DemoProducer");
????????props.put("key.serializer",?"org.apache.kafka.common.serialization.IntegerSerializer");
????????props.put("value.serializer",?"org.apache.kafka.common.serialization.StringSerializer");//創(chuàng)建kafka生產(chǎn)者對象
????????producer?=?new?KafkaProducer<>(props);this.topic?=?topic;this.isAsync?=?isAsync;
????}//運行生產(chǎn)者public?void?run()?{int?messageNo?=?1;while?(true)?{//發(fā)送的消息信息:Message_1
????????????String?messageStr?=?"Message_"?+?messageNo;//開始時間long?startTime?=?System.currentTimeMillis();//是否異步if?(isAsync)?{?//?Send?asynchronously//生產(chǎn)者發(fā)送消息?異步發(fā)送
????????????????producer.send(new?ProducerRecord<>(topic,
????????????????????messageNo,
????????????????????messageStr),?new?DemoCallBack(startTime,?messageNo,?messageStr));
????????????}?else?{?//?Send?synchronouslytry?{//同步發(fā)送
????????????????????producer.send(new?ProducerRecord<>(topic,
????????????????????????messageNo,
????????????????????????messageStr)).get();
????????????????????System.out.println("Sent?message:?("?+?messageNo?+?",?"?+?messageStr?+?")");
????????????????}?catch?(InterruptedException?|?ExecutionException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????}
????????????++messageNo;
????????}
????}
}//進行回調(diào)class?DemoCallBack?implements?Callback?{//開始時間private?final?long?startTime;//keyprivate?final?int?key;//消息private?final?String?message;//構(gòu)造函數(shù)public?DemoCallBack(long?startTime,?int?key,?String?message)?{this.startTime?=?startTime;this.key?=?key;this.message?=?message;
????}/**
?????*?A?callback?method?the?user?can?implement?to?provide?asynchronous?handling?of?request?completion.?This?method?will
?????*?be?called?when?the?record?sent?to?the?server?has?been?acknowledged.?Exactly?one?of?the?arguments?will?be
?????*?non-null.
?????*?用戶可以實現(xiàn)以提供對請求完成的異步處理的回調(diào)方法。?確認發(fā)送到服務(wù)器的消息后，將調(diào)用此方法。?確切地說，其中一個參數(shù)將為非null。
?????*?@param?metadata??The?metadata?for?the?record?that?was?sent?(i.e.?the?partition?and?offset).?Null?if?an?error
?????*??????????????????occurred.
?????*??????????????????發(fā)送的消息的元數(shù)據(jù)(即分區(qū)和偏移量)。?如果發(fā)生錯誤，則為Null。
?????*?@param?exception?The?exception?thrown?during?processing?of?this?record.?Null?if?no?error?occurred.
?????*???????????????????處理此消息期間引發(fā)的異常。?如果沒有發(fā)生錯誤，則為Null。
?????*/public?void?onCompletion(RecordMetadata?metadata,?Exception?exception)?{//經(jīng)過的時間long?elapsedTime?=?System.currentTimeMillis()?-?startTime;//如果元數(shù)據(jù)不為空，則返回null//打印消息發(fā)送到的分區(qū)和偏移量在經(jīng)過的時間if?(metadata?!=?null)?{
????????????System.out.println("message("?+?key?+?",?"?+?message?+?")?sent?to?partition("?+?metadata.partition()?+"),?"?+"offset("?+?metadata.offset()?+?")?in?"?+?elapsedTime?+?"?ms");
????????}?else?{
????????????exception.printStackTrace();
????????}
????}
}

send方法中做了兩件事，一是對消息進行發(fā)送攔截，進行增強，同時進行消息發(fā)送。可以看到在send方法中有大段注釋，而無疑這些注釋是值得我們?nèi)プx的。這里我查看的是異步的方法，通常異步的話，會進行回調(diào)。

這里大段的注釋的大意：這里提到了消息發(fā)送和回調(diào)的方式，同時提到了存儲中偏移量和分區(qū)，提到了事務(wù)、冪等，同時對事務(wù)進行了詳細的介紹和在如果目標主題的消息格式未升級為0.11.0.0，則冪等和事務(wù)性生產(chǎn)請求將失敗。并出現(xiàn){@link org.apache.kafka.common.errors.UnsupportedForMessageFormatException}錯誤。如果在事務(wù)處理期間遇到此問題，則可以中止并繼續(xù)。但是請注意，將來發(fā)送到同一主題的消息將繼續(xù)收到相同的異常，直到升級該主題為止。同時對于阻塞和計算量大的方法需要自己實現(xiàn)線程池的并行。

下面是它的詳細解釋：

1.異步發(fā)送消息到主題，并在確認發(fā)送后調(diào)用提供的回調(diào)。
2.發(fā)送是異步的，并且一旦消息已存儲在等待發(fā)送的消息緩沖區(qū)中，此方法將立即返回。這允許并行發(fā)送許多消息，而不會阻塞等待每條消息之后的響應(yīng)。
3.發(fā)送的結(jié)果是{@link?RecordMetadata}，指定消息發(fā)送到的分區(qū)，分配的偏移量和消息的時間戳。
???如果主題使用{@link?org.apache.kafka.common.record.TimestampType＃CREATE_TIME CreateTime}，則該時間戳記??將是用戶提供的時間戳記，或者如果用戶未為該消息指定時間戳記則是消息發(fā)送時間。
???如果將{@link?org.apache.kafka.common.record.TimestampType＃LOG_APPEND_TIME LogAppendTime}用作主題，則時間戳將是附加消息時的Kafka代理本地時間。
4.由于send調(diào)用是異步的，因此它將為將分配給該消息的{@link?RecordMetadata}返回一個{@link?java.util.concurrent.Future Future}。在此將來調(diào)用{@link?java.util.concurrent.Future＃get()get()}將會阻塞，直到關(guān)聯(lián)的請求完成，然后返回消息的元數(shù)據(jù)或引發(fā)在發(fā)送消息時發(fā)生的任何異常。???
5.如果要模擬一個簡單的阻塞調(diào)用，則可以立即調(diào)用get()方法
6.完全無阻塞的用法可以利用{@link?Callback}參數(shù)來提供將在請求完成后調(diào)用的回調(diào)
7.當用作事務(wù)的一部分時，不必為了檢測 send  code>中的錯誤而定義回調(diào)或檢查將來的結(jié)果。
???如果任何發(fā)送調(diào)用失敗并出現(xiàn)不可恢復的錯誤，則最后一個{@link?#commitTransaction()}調(diào)用將失敗，并從上次失敗的發(fā)送中引發(fā)異常。
???發(fā)生這種情況時，您的應(yīng)用程序應(yīng)調(diào)用{@link?#abortTransaction()}以重置狀態(tài)并繼續(xù)發(fā)送數(shù)據(jù)。
8.某些事務(wù)發(fā)送錯誤無法通過調(diào)用{@link?#abortTransaction()}來解決。特別是，如果事務(wù)發(fā)送以{@link?ProducerFencedException}，{@?link?org.apache.kafka.common.errors.OutOfOrderSequenceException}，
{@?link?org.apache.kafka.common.errors.UnsupportedVersionException}結(jié)尾，?或{@link?org.apache.kafka.common.errors.AuthorizationException}，那么剩下的唯一選擇就是調(diào)用{@link?#close()}。
???致命錯誤導致生產(chǎn)者進入已失效狀態(tài)，在這種狀態(tài)下，將來的API調(diào)用將繼續(xù)引發(fā)包裹在新{@link?KafkaException}中的相同的下標錯誤。???
9.這與啟用冪等性但未配置 transactional.id  code>時相似。在這種情況下，{@ link org.apache.kafka.common.errors.UnsupportedVersionException}和{@link?org.apache.kafka.common.errors.AuthorizationException}被視為致命錯誤。但是，不需要處理{@link?ProducerFencedException}。?此外，可以繼續(xù)在收到{@link?org.apache.kafka.common.errors.OutOfOrderSequenceException}之后發(fā)送消息，但這樣做可能導致未決消息的發(fā)送順序混亂。?為了確保正確的訂閱，您應(yīng)該關(guān)閉生產(chǎn)者并創(chuàng)建一個新實例。????
10.如果目標主題的消息格式未升級為0.11.0.0，則冪等和事務(wù)性生產(chǎn)請求將失敗，并出現(xiàn){@link?org.apache.kafka.common.errors.UnsupportedForMessageFormatException}錯誤。如果在事務(wù)處理期間遇到此問題，則可以中止并繼續(xù)。?但是請注意，將來發(fā)送到同一主題的消息將繼續(xù)收到相同的異常，直到升級該主題為止。???
11. 注意，回調(diào)通常將在生產(chǎn)者的I/O線程中執(zhí)行，因此應(yīng)相當快，否則它們將延遲其他線程的消息發(fā)送。如果要執(zhí)行阻塞或計算量大的回調(diào)，建議在回調(diào)主體中使用自己的{@link?java.util.concurrent.Executor}來并行化處理。???

@Override
public?Future?send(ProducerRecord?record,?Callback?callback)?{
????//?intercept?the?record,?which?can?be?potentially?modified;?this?method?does?not?throw?exceptions
????//攔截消息，進行增強
????ProducerRecord?interceptedRecord?=?this.interceptors.onSend(record);//發(fā)送消息return?doSend(interceptedRecord,?callback);
}

攔截器攔截發(fā)送，進行自定義增強操作：

從注釋里面我們可以看到其進行攔截的時候不會拋出異常，因此需要自己去try…catch

public?ProducerRecord?onSend(ProducerRecord?record)?{
????ProducerRecord?interceptRecord?=?record;for?(ProducerInterceptor?interceptor?:?this.interceptors)?{try?{//在發(fā)送中進行攔截
????????????interceptRecord?=?interceptor.onSend(interceptRecord);
????????}?catch?(Exception?e)?{//?do?not?propagate?interceptor?exception,?log?and?continue?calling?other?interceptors//?be?careful?not?to?throw?exception?from?hereif?(record?!=?null)
????????????????log.warn("Error?executing?interceptor?onSend?callback?for?topic:?{},?partition:?{}",?record.topic(),?record.partition(),?e);else
????????????????log.warn("Error?executing?interceptor?onSend?callback",?e);
????????}
????}return?interceptRecord;
}

追加消息攔截器

//發(fā)送消息中攔截
@Override
public?ProducerRecord?onSend(ProducerRecord?record)?{
????//計數(shù)器
????onSendCount++;
????//如果在發(fā)送中拋異常
????if?(throwExceptionOnSend)
????????throw?new?KafkaException("Injected?exception?in?AppendProducerInterceptor.onSend");
???//返回創(chuàng)建的生產(chǎn)者消息
????return?new?ProducerRecord<>(
????????????record.topic(),?record.partition(),?record.key(),?record.value().concat(appendStr));
}

生產(chǎn)者消息：包含的信息主題、時間戳、分區(qū)、k-v、消息頭

public?ProducerRecord(String?topic,?Integer?partition,?Long?timestamp,?K?key,?V?value,?Iterable?headers)?{
????//主題為空，拋異常
????if?(topic?==?null)
????????throw?new?IllegalArgumentException("Topic?cannot?be?null.");
????//時間戳不為空，或者時間戳小于0，則拋異常
????if?(timestamp?!=?null?&&?timestamp?0)
????????throw?new?IllegalArgumentException(
????????????????String.format("Invalid?timestamp:?%d.?Timestamp?should?always?be?non-negative?or?null.",?timestamp));
????//分區(qū)不為空，或者分區(qū)小于0，則拋異常
????if?(partition?!=?null?&&?partition?0)
????????throw?new?IllegalArgumentException(
????????????????String.format("Invalid?partition:?%d.?Partition?number?should?always?be?non-negative?or?null.",?partition));
????//主題topic
????this.topic?=?topic;
????//分區(qū)
????this.partition?=?partition;
????//key
????this.key?=?key;
????//值
????this.value?=?value;
????//時間戳
????this.timestamp?=?timestamp;
????//消息頭
????this.headers?=?new?RecordHeaders(headers);
}

doSend是我們需要關(guān)注的重點：

?/**
?????*?Implementation?of?asynchronously?send?a?record?to?a?topic.
?????*?實現(xiàn)的異步發(fā)送的記錄到一個主題中
?????*/
????private?Future?doSend(ProducerRecord?record,?Callback?callback)?{
????????TopicPartition?tp?=?null;
????????try?{
????????????//?first?make?sure?the?metadata?for?the?topic?is?available
????????????//首先確保元數(shù)據(jù)提供給topic是可用的??也即準備元數(shù)據(jù)階段
????????????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);
????????????}
????????????//分區(qū)
????????????int?partition?=?partition(record,?serializedKey,?serializedValue,?cluster);
????????????tp?=?new?TopicPartition(record.topic(),?partition);

????????????//將消息頭轉(zhuǎn)成數(shù)組
????????????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
????????????//生產(chǎn)者回調(diào)將確保同時調(diào)用“回調(diào)”和攔截器回調(diào)
????????????Callback?interceptCallback?=?new?InterceptorCallback<>(callback,?this.interceptors,?tp);

????????????//如果事務(wù)管理不為空同時是有事務(wù)的，則添加事務(wù)
????????????if?(transactionManager?!=?null?&&?transactionManager.isTransactional())
????????????????transactionManager.maybeAddPartitionToTransaction(tp);

????????????//在消息收集器中追加信息
????????????RecordAccumulator.RecordAppendResult?result?=?accumulator.append(tp,?timestamp,?serializedKey,
????????????????????serializedValue,?headers,?interceptCallback,?remainingWaitMs);
????????????//如果結(jié)果為空，則sender喚醒??重點
????????????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)?{
????????????//api接口異常
????????????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)?{
????????????//緩沖區(qū)耗盡異常
????????????this.errors.record();
????????????this.metrics.sensor("buffer-exhausted-records").record();
????????????this.interceptors.onSendError(record,?tp,?e);
????????????throw?e;
????????}?catch?(KafkaException?e)?{
????????????//kafka異常
????????????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;
????????}
????}

元數(shù)據(jù)準備階段

?*?Wait?for?cluster?metadata?including?partitions?for?the?given?topic?to?be?available.
?*?等待集群元數(shù)據(jù)包括給定主題的分區(qū)可用。
?*?@param?topic?The?topic?we?want?metadata?for
?*?@param?partition?A?specific?partition?expected?to?exist?in?metadata,?or?null?if?there's?no?preference
?*?@param?maxWaitMs?The?maximum?time?in?ms?for?waiting?on?the?metadata
?*?@return?The?cluster?containing?topic?metadata?and?the?amount?of?time?we?waited?in?ms
?*/
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
????//添加主題
????metadata.add(topic);
????//獲取當前的集群信息而不會阻塞
????Cluster?cluster?=?metadata.fetch();
????//統(tǒng)計分區(qū)數(shù)
????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
????//返回緩存的元數(shù)據(jù)(如果有)，并且記錄的分區(qū)未定義或在已知分區(qū)范圍內(nèi)
????if?(partitionsCount?!=?null?&&?(partition?==?null?||?partition?????????return?new?ClusterAndWaitTime(cluster,?0);

????//開始時間
????long?begin?=?time.milliseconds();
????//記錄等待時間
????long?remainingWaitMs?=?maxWaitMs;
????long?elapsed;
????//?Issue?metadata?requests?until?we?have?metadata?for?the?topic?or?maxWaitTimeMs?is?exceeded.
????//?In?case?we?already?have?cached?metadata?for?the?topic,?but?the?requested?partition?is?greater
????//?than?expected,?issue?an?update?request?only?once.?This?is?necessary?in?case?the?metadata
????//?is?stale?and?the?number?of?partitions?for?this?topic?has?increased?in?the?meantime.
????/**
?????*發(fā)出元數(shù)據(jù)請求，直到超過該主題的元數(shù)據(jù)或maxWaitTimeMs。如果我們已經(jīng)為該主題緩存了元數(shù)據(jù)，
?????*但是請求的分區(qū)大于預期，則僅發(fā)出一次更新請求。
?????*如果元數(shù)據(jù)過時并且與此主題相關(guān)的分區(qū)數(shù)量同時增加，則這是必需的。
?????*/
????do?{
????????log.trace("Requesting?metadata?update?for?topic?{}.",?topic);
????????//添加主題
????????metadata.add(topic);
????????//拿到版本
????????int?version?=?metadata.requestUpdate();
????????//喚醒sender
????????sender.wakeup();
????????try?{
????????????//等待更新
????????????metadata.awaitUpdate(version,?remainingWaitMs);
????????}?catch?(TimeoutException?ex)?{
????????????//?Rethrow?with?original?maxWaitMs?to?prevent?logging?exception?with?remainingWaitMs
????????????throw?new?TimeoutException("Failed?to?update?metadata?after?"?+?maxWaitMs?+?"?ms.");
????????}
????????//獲取集群數(shù)據(jù)
????????cluster?=?metadata.fetch();
????????//計算時間
????????elapsed?=?time.milliseconds()?-?begin;
????????if?(elapsed?>=?maxWaitMs)
????????????throw?new?TimeoutException("Failed?to?update?metadata?after?"?+?maxWaitMs?+?"?ms.");
????????if?(cluster.unauthorizedTopics().contains(topic))
????????????throw?new?TopicAuthorizationException(topic);
????????remainingWaitMs?=?maxWaitMs?-?elapsed;
????????partitionsCount?=?cluster.partitionCountForTopic(topic);
????}?while?(partitionsCount?==?null);

????if?(partition?!=?null?&&?partition?>=?partitionsCount)?{
????????throw?new?KafkaException(
????????????????String.format("Invalid?partition?given?with?record:?%d?is?not?in?the?range?[0...%d).",?partition,?partitionsCount));
????}

????return?new?ClusterAndWaitTime(cluster,?elapsed);
}

集群和等待時間

//集群和等待時間
private?static?class?ClusterAndWaitTime?{
????//集群
????final?Cluster?cluster;
????//等待在元數(shù)據(jù)的時間
????final?long?waitedOnMetadataMs;
????ClusterAndWaitTime(Cluster?cluster,?long?waitedOnMetadataMs)?{
????????this.cluster?=?cluster;
????????this.waitedOnMetadataMs?=?waitedOnMetadataMs;
????}
}

這里值得我們關(guān)注的sender:

sender相關(guān)變量和構(gòu)造函數(shù)

public?class?Sender?implements?Runnable?{
???private?final?Logger?log;

????/*?the?state?of?each?nodes?connection?*/
????//kafka客戶端??每個節(jié)點連接的狀態(tài)
????private?final?KafkaClient?client;

????/*?the?record?accumulator?that?batches?records?*/
????//消息收集器?批量消息
????private?final?RecordAccumulator?accumulator;

????/*?the?metadata?for?the?client?*/
????//元數(shù)據(jù)
????private?final?Metadata?metadata;

????/*?the?flag?indicating?whether?the?producer?should?guarantee?the?message?order?on?the?broker?or?not.?*/
????//生產(chǎn)者是否應(yīng)保證broker上的消息順序的標志
????private?final?boolean?guaranteeMessageOrder;

????/*?the?maximum?request?size?to?attempt?to?send?to?the?server?*/
????//嘗試發(fā)送到服務(wù)器的最大請求大小
????private?final?int?maxRequestSize;

????/*?the?number?of?acknowledgements?to?request?from?the?server?*/
????//要從服務(wù)器請求的確認數(shù)
????private?final?short?acks;

????/*?the?number?of?times?to?retry?a?failed?request?before?giving?up?*/
????//放棄之前重試失敗請求的次數(shù)
????private?final?int?retries;

????/*?the?clock?instance?used?for?getting?the?time?*/
????//用于獲取時間的時鐘實例
????private?final?Time?time;

????/*?true?while?the?sender?thread?is?still?running?*/
????//當發(fā)送方線程仍在運行時為true
????private?volatile?boolean?running;

????/*?true?when?the?caller?wants?to?ignore?all?unsent/inflight?messages?and?force?close.??*/
????//當caller想忽略所有未發(fā)送/正在進行的消息并強制關(guān)閉時為true
????private?volatile?boolean?forceClose;

????/*?metrics?*/
????//發(fā)送的度量信息?相關(guān)指標
????private?final?SenderMetrics?sensors;

????/*?the?max?time?to?wait?for?the?server?to?respond?to?the?request*/
????//等待服務(wù)器響應(yīng)請求的最長時間
????private?final?int?requestTimeout;

????/*?The?max?time?to?wait?before?retrying?a?request?which?has?failed?*/
????//重試失敗的請求之前等待的最長時間
????private?final?long?retryBackoffMs;

????/*?current?request?API?versions?supported?by?the?known?brokers?*/
????//已知broker支持的當前請求API版本
????private?final?ApiVersions?apiVersions;

????/*?all?the?state?related?to?transactions,?in?particular?the?producer?id,?producer?epoch,?and?sequence?numbers?*/
????//與事務(wù)相關(guān)的所有狀態(tài)，特別是生產(chǎn)者ID，生產(chǎn)者時期和序列號
????private?final?TransactionManager?transactionManager;

??????//構(gòu)造函數(shù)
????public?Sender(LogContext?logContext,
??????????????????KafkaClient?client,
??????????????????Metadata?metadata,
??????????????????RecordAccumulator?accumulator,boolean?guaranteeMessageOrder,int?maxRequestSize,short?acks,int?retries,
??????????????????SenderMetricsRegistry?metricsRegistry,
??????????????????Time?time,int?requestTimeout,long?retryBackoffMs,
??????????????????TransactionManager?transactionManager,
??????????????????ApiVersions?apiVersions)?{
????????//通過日志上下文拿到日志信息
????????this.log?=?logContext.logger(Sender.class);
????????//客戶端
????????this.client?=?client;
????????//消息收集器
????????this.accumulator?=?accumulator;
????????//元數(shù)據(jù)
????????this.metadata?=?metadata;
????????//保證消息有序
????????this.guaranteeMessageOrder?=?guaranteeMessageOrder;
????????//最大請求大小
????????this.maxRequestSize?=?maxRequestSize;
????????//運行
????????this.running?=?true;
????????//acks
????????this.acks?=?acks;
????????//重試次數(shù)
????????this.retries?=?retries;
????????//時間
????????this.time?=?time;
????????//發(fā)送指標信息
????????this.sensors?=?new?SenderMetrics(metricsRegistry);
????????//請求超時時間
????????this.requestTimeout?=?requestTimeout;
????????//重試間隔的最短時間
????????this.retryBackoffMs?=?retryBackoffMs;
????????//api版本信息
????????this.apiVersions?=?apiVersions;
????????//事務(wù)管理器
????????this.transactionManager?=?transactionManager;
????}

}

run方法：

/**
?*?The?main?run?loop?for?the?sender?thread
?*?發(fā)送消息線程的主運行循環(huán)
?*/
public?void?run()?{
????log.debug("Starting?Kafka?producer?I/O?thread.");

????//?main?loop,?runs?until?close?is?called
????while?(running)?{
????????try?{
????????????//執(zhí)行run方法?重點
????????????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?{
?????????????//執(zhí)行run方法?重點
????????????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.");
}

執(zhí)行發(fā)送數(shù)據(jù)

/**
?*?Run?a?single?iteration?of?sending
?*?運行一次發(fā)送
?*?@param?now?The?current?POSIX?time?in?milliseconds
?*/
void?run(long?now)?{
????//如果事務(wù)不為空，則放入事務(wù)信息
????if?(transactionManager?!=?null)?{
????????try?{
????????????if?(transactionManager.shouldResetProducerStateAfterResolvingSequences())
????????????????//?Check?if?the?previous?run?expired?batches?which?requires?a?reset?of?the?producer?state.
????????????????//重置生產(chǎn)者id
????????????????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);
????????}
????}

????//發(fā)送生產(chǎn)者數(shù)據(jù)??重點
????long?pollTimeout?=?sendProducerData(now);
????//執(zhí)行poll輪詢操作?進行讀取和寫入操作
????client.poll(pollTimeout,?now);
}

此時進入重要方法sendProduerData()方法和poll()方法。

總結(jié)

以上是生活随笔為你收集整理的kafka 异步发送阻塞_Kafka学习一的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。