前言: kafka是一個輕量級的/分散式的/具備replication能力的日誌採集元件,通常被整合到應用系統中,收集"使用者行為日誌"等,並可以使用各種消費終端(consumer)將訊息轉存到HDFS等其他結構化資料儲存系統中.因為日誌訊息通常為文字資料,尺寸較小,且對實時性以及資料可靠性要求不嚴格,但是需要日誌儲存端具備較高的資料吞吐能力,這種"寬鬆"的設計要求,非常適合使用kafka. 

一.入門

    1.1 簡介

    Kafka是一個"分散式的"/"可分割槽的(partitioned)"/"基於備份的(replicated)"/"基於commit-log儲存"的服務. 它提供了類似於JMS的特性,但是在設計實現上完全不同,此外它並不是JMS規範的實現.

    kafka訊息是根據Topic進行歸類,傳送訊息者成為Producer,訊息接收者成為Consumer;此外kafka叢集有多個kafka例項組成,每個例項(server)稱為broker.

    無論是kafka叢集,還是producer和consumer都依賴於zookeeper來保證系統可用性以及儲存一些meta資訊.

 

(摘自官網) 

    其中client與server的通訊,都是基於TCP,而且訊息協議非常輕量級.

    Topics/logs

    一個Topic可以認為是一類訊息,每個topic將被分成多個partition(區),每個partition在儲存層面是append log檔案.任何釋出到此partition的訊息都會直接追加到log檔案的尾部,每條訊息在檔案中的位置稱為offset(偏移量),offset為一個long型數字,它唯一的標記一條訊息.kafka並沒有提供其他額外的索引機制來儲存offset,因為在kafka中幾乎不允許對訊息進行"隨機讀-寫",一旦訊息寫入log日誌之後,將不能被修改.

(摘自官網) 

    kafka和JMS實現(activeMQ)不同的是:即使訊息被消費,訊息仍然不會被立即刪除.日誌檔案將會根據broker中的配置要求,保留一定的時間之後刪除;比如log檔案保留2天,那麼兩天後,檔案會被清除,無論其中的訊息是否被消費.kafka通過這種簡單的手段,來釋放磁碟空間.此外,kafka的效能並不會因為日誌檔案的太多而低下,所以即使保留較多的log檔案,也不不會有問題.

    對於consumer而言,它需要儲存消費訊息的offset,對於offset的儲存和使用,有consumer來控制;當consumer正常消費訊息時,offset將會"線性"的向前驅動,即訊息將依次順序被消費.事實上consumer可以使用任意順序消費訊息,它只需要將offset重置為任意值..(offset將會儲存在zookeeper中,參見下文)

    kafka叢集幾乎不需要維護任何consumer和producer狀態資訊,這些資訊有zookeeper儲存;因此producer和consumer的客戶端實現非常輕量級,它們可以隨意離開,而不會對叢集造成額外的影響.

    partitions的設計目的有多個.最根本原因是kafka基於檔案儲存.通過分割槽,可以將日誌內容分散到多個server上,來避免檔案尺寸達到單機磁碟的上限,每個partiton都會被當前server(kafka例項)儲存;可以將一個topic切分多任意多個partitions(備註:基於sharding),來訊息儲存/消費的效率.此外越多的partitions意味著可以容納更多的consumer,有效提升併發消費的能力.(具體原理參見下文).

    Distribution

    一個Topic的多個partitions,被分佈在kafka叢集中的多個server上;每個server(kafka例項)負責partitions中訊息的讀寫操作;此外kafka還可以配置每個partition需要備份的個數(replicas),每個partition將會被備份到多臺機器上,以提高可用性.[replicas特性在0.8V才支援]

    基於replicated方案,那麼就意味著需要對多個備份進行排程;一個partition可以在多個server上備份,那麼其中一個server作為此partiton的leader;leader負責此partition所有的讀寫操作,如果leader失效,那麼將會有其他follower來接管(成為新的leader);follower只是單調的和leader跟進,同步訊息即可..由此可見作為leader的server承載了全部的請求壓力,因此從叢集的整體考慮,有多少個partitions就意味著有多少個"leader",kafka會將"leader"均衡的分散在每個例項上,來確保整體的效能穩定.[備註:kafka中將leader角色許可權下放到partition這個層級]

 

kafka-cluster 

    Producers

    Producer將訊息釋出到指定的Topic中,同時Producer也能決定將此訊息傳送到哪個partition;如果一個Topic有多個partitions時,你需要選擇partition是演算法,比如基於"round-robin"方式或者通過其他的一些演算法等.無論如何選擇partition路由演算法,我們最直接的目的就是希望訊息能夠均勻的傳送給每個partition,這樣可以讓consumer消費的訊息量也能"均衡".

    Consumers

    本質上kafka只支援Topic.每個consumer屬於一個consumer group;反過來說,每個group中可以有多個consumer.對於Topic中的一條特定的訊息,只會被訂閱此Topic的每個group中的一個consumer消費,此訊息不會發送給一個group的多個consumer;那麼一個group中所有的consumer將會交錯的消費整個Topic.

    如果所有的consumer都具有相同的group,這種情況和JMS queue模式很像;訊息將會在consumers之間負載均衡.

    如果所有的consumer都具有不同的group,那這就是"釋出-訂閱";訊息將會廣播給所有的消費者.

(摘自官網) 

    在kafka中,一個partition中的訊息只會被group中的一個consumer消費(同一時刻);每個group中consumer訊息消費互相獨立;我們可以認為一個group是一個"訂閱"者,一個Topic中的每個partions,只會被一個"訂閱者"中的一個consumer消費,不過一個consumer可以同時消費多個partitions中的訊息.kafka只能保證一個partition中的訊息被某個consumer消費時是順序的.事實上,從Topic角度來說,當有多個partitions時,訊息仍不是全域性有序的.

    通常情況下,一個group中會包含多個consumer,這樣不僅可以提高topic中訊息的併發消費能力,而且還能提高"故障容錯"性,如果group中的某個consumer失效,那麼其消費的partitions將會有其他consumer自動接管.

    kafka的設計原理決定,對於一個topic,同一個group中不能有多於partitions個數的consumer同時消費,否則將意味著某些consumer將無法得到訊息.

    Guarantees

    1) 傳送到partitions中的訊息將會按照它接收的順序追加到日誌中,無論一個partition由多少個log檔案構成,那麼它傳送給consumer的順序是一定的.

    2) 對於消費者而言,它們消費訊息的順序和日誌中訊息順序一致.

    3) 如果Topic的"replication factor"為N,那麼允許N-1個kafka例項失效.只要有一個replication存活,那麼此partition的讀寫操作都不會中斷.

    1.2 Use cases

    Messaging

    和一些常規的訊息系統相比,kafka仍然是個不錯的選擇;它具備partitons/replication和容錯,可以使kafka具有良好的擴充套件性和效能優勢.不過到目前為止,我們應該很清楚認識到,kafka並沒有提供JMS中的"事務性""訊息傳輸擔保(訊息確認機制)""訊息分組"等企業級特性;kafka只能使用作為"常規"的訊息系統,在一定程度上,尚未確保訊息的傳送與接收絕對可靠(比如,訊息重發,訊息傳送丟失等)

    Websit activity tracking

    kafka可以作為"網站活性跟蹤"的最佳工具;可以將網頁/使用者操作等資訊傳送到kafka中.並實時監控,或者離線統計分析等.

    Log Aggregation

    kafka的特性決定它非常適合作為"日誌收集中心";application可以將操作日誌"批量""非同步"的傳送到kafka叢集中,而不是儲存在本地或者DB中;kafka可以批量提交訊息/壓縮訊息等,這對producer端而言,幾乎感覺不到效能的開支.此時consumer端可以使hadoop等其他系統化的儲存和分析系統.

二. 設計原理

    kafka的設計初衷是希望做為一個統一的資訊收集平臺,能夠實時的收集反饋資訊,並需要能夠支撐較大的資料量,且具備良好的容錯能力.

    1.Persistence

    kafka使用檔案儲存訊息(append only log),這就直接決定kafka在效能上嚴重依賴檔案系統的本身特性.且無論任何OS下,對檔案系統本身的優化是非常艱難的.檔案快取/直接記憶體對映等是常用的手段.因為kafka是對日誌檔案進行append操作,因此磁碟檢索的開支是較小的;同時為了減少磁碟寫入的次數,broker會將訊息暫時buffer起來,當訊息的個數(或尺寸)達到一定閥值時,再flush到磁碟,這樣減少了磁碟IO呼叫的次數.對於kafka而言,較高效能的磁碟,將會帶來更加直接的效能提升.

    2.Efficiency

    需要考慮的影響效能點很多,除磁碟IO之外,我們還需要考慮網路IO,這直接關係到kafka的吞吐量問題.kafka並沒有提供太多高超的技巧;對於producer端,可以將訊息buffer起來,當訊息的條數達到一定閥值時,批量傳送給broker;對於consumer端也是一樣,批量fetch多條訊息.不過訊息量的大小可以通過配置檔案來指定.對於kafka broker端,似乎有個sendfile系統呼叫可以潛在的提升網路IO的效能:將檔案的資料對映到系統記憶體中,socket直接讀取相應的記憶體區域即可,而無需程序再次copy和交換(這裡涉及到"磁碟IO資料"/"核心記憶體"/"程序記憶體"/"網路緩衝區",多者之間的資料copy).

    其實對於producer/consumer/broker三者而言,CPU的開支應該都不大,因此啟用訊息壓縮機制是一個良好的策略;壓縮需要消耗少量的CPU資源,不過對於kafka而言,網路IO更應該需要考慮.可以將任何在網路上傳輸的訊息都經過壓縮.kafka支援gzip/snappy等多種壓縮方式.

    3. Producer

    Load balancing

    kafka叢集中的任何一個broker,都可以向producer提供metadata資訊,這些metadata中包含"叢集中存活的servers列表"/"partitions leader列表"等資訊(請參看zookeeper中的節點資訊). 當producer獲取到metadata信心之後, producer將會和Topic下所有partition leader保持socket連線;訊息由producer直接通過socket傳送到broker,中間不會經過任何"路由層".事實上,訊息被路由到哪個partition上,有producer客戶端決定.比如可以採用"random""key-hash""輪詢"等,如果一個topic中有多個partitions,那麼在producer端實現"訊息均衡分發"是必要的.在producer端的配置檔案中,開發者可以指定partition路由的方式.

    Asynchronous send

    將多條訊息暫且在客戶端buffer起來,並將他們批量傳送到broker;小資料IO太多,會拖慢整體的網路延遲,批量延遲傳送事實上提升了網路效率;不過這也有一定的隱患,比如當producer失效時,那些尚未傳送的訊息將會丟失.

    4.Consumer

    consumer端向broker傳送"fetch"請求,並告知其獲取訊息的offset;此後consumer將會獲得一定條數的訊息;consumer端也可以重置offset來重新消費訊息.[備註:offset,訊息偏移量,integer值,broker可以根據offset來決定訊息的起始位置]

    在JMS實現中,Topic模型基於push方式,即broker將訊息推送給consumer端.不過在kafka中,採用了pull方式,即consumer在和broker建立連線之後,主動去pull(或者說fetch)訊息;這中模式有些優點,首先consumer端可以根據自己的消費能力適時的去fetch訊息並處理,且可以控制訊息消費的進度(offset);此外,消費者可以良好的控制訊息消費的數量,batch fetch.

    其他JMS實現,訊息消費的位置是有prodiver保留,以便避免重複傳送訊息或者將沒有消費成功的訊息重發等,同時還要控制訊息的狀態.這就要求JMS broker需要太多額外的工作.在kafka中,partition中的訊息只有一個consumer在消費,且不存在訊息狀態的控制,也沒有複雜的訊息確認機制,可見kafka broker端是相當輕量級的.當訊息被consumer接收之後,consumer可以在本地儲存最後訊息的offset,並間歇性的向zookeeper註冊offset.由此可見,consumer客戶端也很輕量級.

    這就意味著,kafka中consumer負責維護訊息的消費記錄,而broker則不關心這些,這種設計不僅提高了consumer端的靈活性,也適度的減輕了broker端設計的複雜度;這是和眾多JMS prodiver的區別.此外,kafka中訊息ACK的設計也和JMS有很大不同,kafka中的訊息時批量(通常以訊息的條數或者chunk的尺寸為單位)傳送給consumer,當訊息消費成功後,向zookeeper提交訊息的offset,而不會向broker交付ACK.或許你已經意識到,這種"寬鬆"的設計,將會有"丟失"訊息/"訊息重發"的危險.

    5.Message Delivery Semantics

    對於JMS實現,訊息傳輸擔保非常直接:有且只有一次(exactly once).在kafka中稍有不同,對於consumer而言:

    1) at most once: 最多一次,這個和JMS中"非持久化"訊息類似.傳送一次,無論成敗,將不會重發.

    2) at least once: 訊息至少傳送一次,如果訊息未能接受成功,可能會重發,直到接收成功.

    3) exactly once: 訊息只會傳送一次.

    at most once: 消費者fetch訊息,然後儲存offset,然後處理訊息;當client儲存offset之後,但是在訊息處理過程中consumer程序失效(crash),導致部分訊息未能繼續處理.那麼此後可能其他consumer會接管,但是因為offset已經提前儲存,那麼新的consumer將不能fetch到offset之前的訊息(儘管它們尚沒有被處理),這就是"at most once".

    at least once: 消費者fetch訊息,然後處理訊息,然後儲存offset.如果訊息處理成功之後,但是在儲存offset階段zookeeper異常或者consumer失效,導致儲存offset操作未能執行成功,這就導致接下來再次fetch時可能獲得上次已經處理過的訊息,這就是"at least once".

    exactly once: kafka中並沒有嚴格的去實現(基於2階段提交,事務),我們認為這種策略在kafka中是沒有必要的.

    因為"訊息消費"和"儲存offset"這兩個操作的先後時機不同,導致了上述3種情況,通常情況下"at-least-once"是我們搜選.(相比at most once而言,重複接收資料總比丟失資料要好).

 

    6. Replication

    kafka中,replication策略是基於partition,而不是topic;kafka將每個partition資料複製到多個server上,任何一個partition有一個leader和多個follower(可以沒有);備份的個數可以通過broker配置檔案來設定.leader處理所有的read-write請求,follower需要和leader保持同步.Follower就像一個"consumer",消費訊息並儲存在本地日誌中;leader負責跟蹤所有的follower狀態,如果follower"落後"太多或者失效,leader將會把它從replicas同步列表中刪除.當所有的follower都將一條訊息儲存成功,此訊息才被認為是"committed",那麼此時consumer才能消費它,這種同步策略,就要求follower和leader之間必須具有良好的網路環境.即使只有一個replicas例項存活,仍然可以保證訊息的正常傳送和接收,只要zookeeper叢集存活即可.(備註:不同於其他分散式儲存,比如hbase需要"多數派"存活才行)

    kafka判定一個follower存活與否的條件有2個:1) follower需要和zookeeper保持良好的連結 2) 它必須能夠及時的跟進leader,不能落後太多.如果同時滿足上述2個條件,那麼leader就認為此follower是"活躍的".如果一個follower失效(server失效)或者落後太多,leader將會把它從同步列表中移除[備註:如果此replicas落後太多,它將會繼續從leader中fetch資料,直到足夠up-to-date,然後再次加入到同步列表中;kafka不會更換replicas宿主!因為"同步列表"中replicas需要足夠快,這樣才能保證producer釋出訊息時接受到ACK的延遲較小].

    當leader失效時,需在followers中選取出新的leader,可能此時follower落後於leader,因此需要選擇一個"up-to-date"的follower.kafka中leader選舉並沒有採用"投票多數派"的演算法,因為這種演算法對於"網路穩定性"/"投票參與者數量"等條件有較高的要求,而且kafka叢集的設計,還需要容忍N-1個replicas失效.對於kafka而言,每個partition中所有的replicas資訊都可以在zookeeper中獲得,那麼選舉leader將是一件非常簡單的事情.選擇follower時需要兼顧一個問題,就是新leader server上所已經承載的partition leader的個數,如果一個server上有過多的partition leader,意味著此server將承受著更多的IO壓力.在選舉新leader,需要考慮到"負載均衡",partition leader較少的broker將會更有可能成為新的leader.

    在整幾個叢集中,只要有一個replicas存活,那麼此partition都可以繼續接受讀寫操作.

    7.Log

    如果一個topic的名稱為"my_topic",它有2個partitions,那麼日誌將會儲存在my_topic_0和my_topic_1兩個目錄中;日誌檔案中儲存了一序列"log entries"(日誌條目),每個log entry格式為"4個位元組的數字N表示訊息的長度" + "N個位元組的訊息內容";每個日誌都有一個offset來唯一的標記一條訊息,offset的值為8個位元組的數字,表示此訊息在此partition中所處的起始位置..每個partition在物理儲存層面,有多個log file組成(稱為segment).segment file的命名為"最小offset".kafka.例如"00000000000.kafka";其中"最小offset"表示此segment中起始訊息的offset.

(摘自官網) 

    其中每個partiton中所持有的segments列表資訊會儲存在zookeeper中.

    當segment檔案尺寸達到一定閥值時(可以通過配置檔案設定,預設1G),將會建立一個新的檔案;當buffer中訊息的條數達到閥值時將會觸發日誌資訊flush到日誌檔案中,同時如果"距離最近一次flush的時間差"達到閥值時,也會觸發flush到日誌檔案.如果broker失效,極有可能會丟失那些尚未flush到檔案的訊息.因為server意外失效,仍然會導致log檔案格式的破壞(檔案尾部),那麼就要求當server啟東是需要檢測最後一個segment的檔案結構是否合法並進行必要的修復.

    獲取訊息時,需要指定offset和最大chunk尺寸,offset用來表示訊息的起始位置,chunk size用來表示最大獲取訊息的總長度(間接的表示訊息的條數).根據offset,可以找到此訊息所在segment檔案,然後根據segment的最小offset取差值,得到它在file中的相對位置,直接讀取輸出即可.

    日誌檔案的刪除策略非常簡單:啟動一個後臺執行緒定期掃描log file列表,把儲存時間超過閥值的檔案直接刪除(根據檔案的建立時間).為了避免刪除檔案時仍然有read操作(consumer消費),採取copy-on-write方式.

    8.Distribution

    kafka使用zookeeper來儲存一些meta資訊,並使用了zookeeper watch機制來發現meta資訊的變更並作出相應的動作(比如consumer失效,觸發負載均衡等)

    1) Broker node registry: 當一個kafka broker啟動後,首先會向zookeeper註冊自己的節點資訊(臨時znode),同時當broker和zookeeper斷開連線時,此znode也會被刪除.

    格式: /broker/ids/[0...N]   -->host:port;其中[0..N]表示broker id,每個broker的配置檔案中都需要指定一個數字型別的id(全域性不可重複),znode的值為此broker的host:port資訊.

    2) Broker Topic Registry: 當一個broker啟動時,會向zookeeper註冊自己持有的topic和partitions資訊,仍然是一個臨時znode.

    格式: /broker/topics/[topic]/[0...N]  其中[0..N]表示partition索引號.

    3) Consumer and Consumer group: 每個consumer客戶端被建立時,會向zookeeper註冊自己的資訊;此作用主要是為了"負載均衡".

    一個group中的多個consumer可以交錯的消費一個topic的所有partitions;簡而言之,保證此topic的所有partitions都能被此group所消費,且消費時為了效能考慮,讓partition相對均衡的分散到每個consumer上.

    4) Consumer id Registry: 每個consumer都有一個唯一的ID(host:uuid,可以通過配置檔案指定,也可以由系統生成),此id用來標記消費者資訊.

    格式: /consumers/[group_id]/ids/[consumer_id]

    仍然是一個臨時的znode,此節點的值為{"topic_name":#streams...},即表示此consumer目前所消費的topic + partitions列表.

    5) Consumer offset Tracking: 用來跟蹤每個consumer目前所消費的partition中最大的offset.

    格式: /consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]   -->offset_value

    此znode為持久節點,可以看出offset跟group_id有關,以表明當group中一個消費者失效,其他consumer可以繼續消費.

    6) Partition Owner registry: 用來標記partition正在被哪個consumer消費.臨時znode

    格式: /consumers/[group_id]/owners/[topic]/[broker_id-partition_id]   -->consumer_node_id

    此節點表達了"一個partition"只能被group下一個consumer消費,同時當group下某個consumer失效,那麼將會觸發負載均衡(即:讓partitions在多個consumer間均衡消費,接管那些"遊離"的partitions)

    當consumer啟動時,所觸發的操作:

    A) 首先進行"Consumer id Registry";

    B) 然後在"Consumer id Registry"節點下注冊一個watch用來監聽當前group中其他consumer的"leave"和"join";只要此znode path下節點列表變更,都會觸發此group下consumer的負載均衡.(比如一個consumer失效,那麼其他consumer接管partitions).

    C) 在"Broker id registry"節點下,註冊一個watch用來監聽broker的存活情況;如果broker列表變更,將會觸發所有的groups下的consumer重新balance.

    Consumer均衡演算法

    當一個group中,有consumer加入或者離開時,會觸發partitions均衡.均衡的最終目的,是提升topic的併發消費能力.

     1) 假如topic1,具有如下partitions: P0,P1,P2,P3

     2) 加入group中,有如下consumer: C0,C1

     3) 首先根據partition索引號對partitions排序: P0,P1,P2,P3

     4) 根據consumer.id排序: C0,C1

     5) 計算倍數: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)

     6) 然後依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

    總結: 

    1) Producer端使用zookeeper用來"發現"broker列表,以及和Topic下每個partition leader建立socket連線併發送訊息.

    2) Broker端使用zookeeper用來註冊broker資訊,已經監測partition leader存活性.

    3) Consumer端使用zookeeper用來註冊consumer資訊,其中包括consumer消費的partition列表等,同時也用來發現broker列表,並和partition leader建立socket連線,並獲取訊息.

三.主要配置 

    1.Broker主要配置

Xml程式碼  

1. ##broker標識,cluster中,此ID必須唯一  
2. broker.id=0
3. ##接受consumer/producer的連結埠  
4. port=9092
5. ##用來維護叢集狀態,以及consumer消費記錄  
6. ##consumer和broker必須接入到同一個zk環境中.  
7. zookeeper.connect=localhost:2181  
8. zookeeper.connection.timeout.ms=30000
9. ##broker所能接受的訊息的最大尺寸  
10. ##producer不能釋出更大尺寸的message  
11. messages.max.bytes=1000000
12. ##broker在處理client請求是,允許開啟的執行緒個數.預設為3.  
13. num.network.threads=3
14. ##用於磁碟IO操作的執行緒的個數,預設為8,建議和磁碟的個數保持一致  
15. num.io.threads=8
16. ##允許入隊的最大請求數,"資料操作請求"首先加入佇列,等待IO執行緒  
17. ##進行磁碟操作獲取資料,資料操作結束後,請求被移除佇列並由network  
18. ##執行緒響應給client端.此引數用於控制"等待IO處理的請求數".  
19. queued.max.requests=500
20. #socket調優引數: sendBuffer (SO_SNDBUF)  
21. socket.send.buffer.bytes=1048576
22. ##socket調優引數:receiveBuffer (SO_RCVBUFF)  
23. socket.receive.buffer.bytes=1048576
24. # The maximum size of a request that the socket server will accept (protection against OOM)  
25. socket.request.max.bytes=104857600
26. #################Log##########  
27. log.dirs=/tmp/kafka-logs  
28. ##每個topic的分割槽數.  
29. ##kafka的特點就在於"分割槽",每個Topic被拆分成多個partitions  
30. ##partitions可以被sharding到多個broker上,以提高併發能力和"可用性"  
31. num.partitions=2
32. ##log檔案片段的最大尺寸,每個partition(邏輯上)的資料都會被寫入到磁碟的  
33. ##log檔案中(append only),此引數用於控制單個檔案的大小.  
34. ## 1024*1024*1024,1G  
35. ##log.segment.bytes=  
36. ##log檔案"sync"到磁碟之前累積的訊息條數  
37. ##因為磁碟IO操作是一個慢操作,但又是一個"資料可靠性"的必要手段  
38. ##所以此引數的設定,需要在"資料可靠性"與"效能"之間做必要的權衡.  
39. ##如果此值過大,將會導致每次"fsync"的時間較長(IO阻塞)  
40. ##如果此值過小,將會導致"fsync"的次數較多,這也意味著整體的client請求有一定的延遲.  
41. ##物理server故障,將會導致沒有fsync的訊息丟失.  
42. ##預設值為10000  
43. log.flush.interval.messages=10000
44. ##僅僅通過interval來控制訊息的磁碟寫入時機,是不足的.  
45. ##此引數用於控制"fsync"的時間間隔,如果訊息量始終沒有達到閥值,但是離上一次磁碟同步的時間間隔  
46. ##達到閥值,也將觸發.  
47. log.flush.interval.ms=1000
48. #對某些特定的topic而言,重寫log.flush.interval.messages屬性  
49. ##log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000  
50. ######################  
51. ##是否自動建立topic  
52. ##如果broker中沒有topic的資訊,當producer/consumer操作topic時,是否自動建立.  
53. ##如果為false,則只能通過API或者command建立topic  
54. auto.create.topics.enable=true
55. ##partition leader與replicas之間通訊時,socket的超時時間  
56. controller.socket.timeout.ms=30000
57. ##partition leader與replicas資料同步時,訊息的佇列尺寸.  
58. controller.message.queue.size=10
59. ##partitions的"replicas"個數,不得大於叢集中broker的個數  
60. default.replication.factor=1
61. ##partition Leader和follower通訊時,如果在此時間內,沒有收到follower的"fetch請求"  
62. ##leader將會認為follower"失效",將不會與其同步訊息.[follower主動跟隨leader,並請求同步訊息]  
63. replica.lag.time.max.ms=10000
64. ##如果follower落後與leader太多,將會認為此follower[或者說partition relicas]已經失效  
65. ##通常,在follower與leader通訊時,因為網路延遲或者連結斷開,總會導致replicas中訊息同步滯後  
66. ##如果訊息之後太多,leader將認為此follower網路延遲較大或者訊息吞吐能力有限,將會把此replicas遷移  
67. ##到其他follower中.  
68. ##在broker數量較少,或者網路不足的環境中,建議提高此值.  
69. replica.lag.max.messages=4000
70. ##follower與leader之間的socket超時時間  
71. 相關推薦

    Apache kafka原理與特性(0.8V)

    前言: kafka是一個輕量級的/分散式的/具備replication能力的日誌採集元件,通常被整合到應用系統中,收集"使用者行為日誌"等,並可以使用各種消費終端(consumer)將訊息轉存到HDFS等其他結構化資料儲存系統中.因為日誌訊息通常為文字資料,尺寸較小,且對

    DataPipeline |Apache Kafka實戰作者胡夕：Apache Kafka監控與調優

    推出 充足 不足 交互 進入 時間片 第一條 小時 send 胡夕，《Apache Kafka實戰》作者，北航計算機碩士畢業，現任某互金公司計算平臺總監，曾就職於IBM、搜狗、微博等公司。國內活躍的Kafka代碼貢獻者。 前言雖然目前Apache Kafka已經全面進化成

    學習 Apache httpd 原理與配置

    一、Apache的簡介      Apache  Httpd又可以簡稱為httpd或者Apache，它是Internet使用最廣泛的web伺服器之一。     使用Apache提供的web伺服器是由守護程序 httpd，通過http協議進行文字傳輸，預設使用80埠的明文傳輸方

    Apache Kafka 簡介與使用

    Kafka 可以簡單理解為分散式MQ，用Scala編寫，執行在JVM上。 分散式程式，除了其自身的基本概念外，最重要的就是要知道它是如何實現高併發和高可用的： Kafka 用 Partitions 實現了高併發； Kafka 用 Partitions 複製

    Apache Kafka -7 與Storm整合

    Apache Kafka教程 之 與Storm整合 Apache Kafka - 與Storm整合 關於Storm Storm最初是由Nathan Marz和BackType建立的。在短時間內，Apache Storm成為分散式實時

    Apache Kafka 0.10.0.0&0.11.0.0新特性 更新日誌

    一、About 0.10.0.0 Apache Kafka 0.10.0.0於美國時間2016年5月24日正式釋出。Apache Kafka 0.10.0.0是Apache Kafka的主要版本。以下是新特性： 1、Kafka Streams Kafk

    Apache Kafka 0.11版本新功能簡介

    多個 spa 實現 cer true assign 線程 cto headers Apache Kafka近日推出0.11版本。這是一個裏程碑式的大版本，特別是Kafka從這個版本開始支持“exactly-once”語義(下稱EOS， exactly-once semant

    [轉]Kafka 設計與原理詳解

    RabbitMQ VS Apache Kafka （九）—— RabbitMQ叢集的分割槽容錯性與高可用性

    本章，我們討論有關RabbitMQ的容錯性，訊息一致性及高可用性。RabbitMQ可以作為叢集節點來執行，因此RabbitMQ通常被歸為分散式訊息系統，對於分散式訊息系統，我們的關注點通常是一致性與可用性。 我們為什麼要討論分散式系統的一致性與可用性，本質在於兩者描述的是系統在失敗的

    Apache Kylin 入門 2 - 原理與架構

    Apache Kylin 入門系列目錄 Apache Kylin 入門 1 - 基本概念 Apache Kylin 入門 2 - 原理與架構 工作原理 簡單來說，Kylin 的核心思想是預計算（利用空間換時間），即對多維分析可能用到的度量進行預計算，將計算好的結果儲存成 Cube 並存在

    Apache Kafka核心元件和流程-協調器（消費者和組協調器）-設計-原理（入門教程輕鬆學）

    作者：稀有氣體 來源：CSDN 原文：https://blog.csdn.net/liyiming2017/article/details/82805479 版權宣告：本文為博主原創文章，轉載請附上博文連結！ 本入門教程，涵蓋Kafka核心內容，通過例項和大量圖表，幫助學習

    Apache Kafka入門教程輕鬆學-第四章 Kafka核心元件和流程-設計-原理（二）協調器（消費者和組協調器）

    本入門教程，涵蓋Kafka核心內容，通過例項和大量圖表，幫助學習者理解，任何問題歡迎留言。 目錄： 上一節介紹了kafka工作的核心元件--控制器。本節將介紹消費者密切相關的元件--協調器。它負責消費者的出入組工作。大家可以回想一下kafka核心概念中關於吃蘋果的場景，如

    Apache Kafka入門教程輕鬆學-第四章 Kafka核心元件和流程-設計-原理（一）控制器

    本入門教程，涵蓋Kafka核心內容，通過例項和大量圖表，幫助學習者理解，任何問題歡迎留言。 目錄： 通過前幾章的學習，我們已經從巨集觀層面瞭解了kafka的設計理念。包括kafka叢集的組成、訊息的主題、主題的分割槽、分割槽的副本等內容。接下來我們會繼續深入，瞭解kafk

    Apache Kafka的Exactly-once的定義 原理和實現

    2018年，Apache Kafka以一種特殊的設計和方法實現了強語義的exactly-once和事務性。熱淚盈眶啊！ 這篇文章將講解kafka中exactly-once和事務操作的原理，具體為 （1）exactly-once在kafka中的定義。 （2）資料

    Kafka原理——fabric1.0版本中的節點排序方法

    Kafka原理 可參考Zookeeper一起理解，後續自己在專案中實現，會再來補充一些實踐的內容。 Zookeeper整理:https://blog.csdn.net/yangwei256/article/details/83786917 目錄 概述....  1

    Apache Kafka入門教程輕鬆學-第四章 Kafka核心元件和流程-設計-原理（四）副本管理器

    本入門教程，涵蓋Kafka核心內容，通過例項和大量圖表，幫助學習者理解，任何問題歡迎留言。 目錄： 本章簡單介紹了副本管理器，副本管理器負責分割槽及其副本的管理。副本管理器具體的工作流程可以參考牟大恩所著的《Kafka入門與實踐》。 副本管理器 副本機制使得kafka

    WARN Connection to node 0 could not be established. Broker may not be available. (org.apache.kafka.c

    筆者啟動kafka後提示WARN Connection to node 0 could not be established. Broker may not be available. (org.apache.kafka.c錯誤，網上查了下都沒有解決，最後筆者的解決方法是： 註釋掉liste

    事無鉅細 Apache Kafka 0.9.0.1 叢集環境搭建

    Kafka叢集環境依賴於Zookeeper環境。因此我們的環境搭建實際分為兩部分。Zookeeper環境搭建和Kafka環境搭建。 Zookeeper 3.4.8叢集搭建 部署安裝包 下載 wget http://mirrors.cn

    kafka使用與設計原理

    第一部分 kafka基礎簡介 kafka簡介 kafka是apache開源的基於zookeeper協調的分散式訊息系統，具有高吞吐率（可做到單機每秒幾十萬qps，基於磁碟進行儲存，做到時間複雜度O(1) ）、高效能、實時、高可靠等特點，可實時處理流式資料。最早由Linkedin公司用

    Spark的RDD原理以及2.0特性的介紹

    Spark 是什麼  Spark 是 Apache 頂級專案裡面最火的大資料處理的計算引擎，它目前是負責大資料計算的工作。包括離線計算或互動式查詢、資料探勘演算法、流式計算以及圖計算等。全世界有許多公司和組織使用或給社群貢獻程式碼，社群的活躍度見 www.github.com/apache/spar