前言: 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代碼?? ##broker標識,cluster中,此ID必須唯一??

• ##接受consumer/producer的鏈接端口??
• ##用來維護集群狀態,以及consumer消費記錄??
• zookeeper.connect=localhost:2181??
• ##broker所能接受的消息的最大尺寸??
• messages.max.bytes=1000000??
• num.network.threads=3??
• num.io.threads=8??
• ##進行磁盤操作獲取數據,數據操作結束后,請求被移除隊列并由network??
• queued.max.requests=500??
• socket.send.buffer.bytes=1048576??
• socket.receive.buffer.bytes=1048576??
• socket.request.max.bytes=104857600??
• log.dirs=/tmp/kafka-logs??
• ##kafka的特點就在于"分區",每個Topic被拆分成多個partitions??
• num.partitions=2??
• ##log文件中(append?only),此參數用于控制單個文件的大小.??
• ##log.segment.bytes=??
• ##log文件"sync"到磁盤之前累積的消息條數??
• ##所以此參數的設置,需要在"數據可靠性"與"性能"之間做必要的權衡.??
• ##如果此值過小,將會導致"fsync"的次數較多,這也意味著整體的client請求有一定的延遲.??
• ##默認值為10000??
• ##僅僅通過interval來控制消息的磁盤寫入時機,是不足的.??
• ##達到閥值,也將觸發.??
• #對某些特定的topic而言,重寫log.flush.interval.messages屬性??
• ??
• ##是否自動創建topic??
• ##如果為false,則只能通過API或者command創建topic??
• ##partition?leader與replicas之間通訊時,socket的超時時間??
• ##partition?leader與replicas數據同步時,消息的隊列尺寸.??
• ##partitions的"replicas"個數,不得大于集群中broker的個數??
• ##partition?Leader和follower通訊時,如果在此時間內,沒有收到follower的"fetch請求"??
• replica.lag.time.max.ms=10000??
• ##通常,在follower與leader通訊時,因為網絡延遲或者鏈接斷開,總會導致replicas中消息同步滯后??
• ##到其他follower中.??
• replica.lag.max.messages=4000??
• replica.socket.timeout.ms=30000??
• ##沒有意義的參數??
• ##當follower的fetch請求發出后,等待leader發送數據的時間.??
• replica.fetch.wait.max.ms=500??
• replica.fetch.min.bytes=1??
• num.replica.fetchers=1??
• ##檢測log文件的時間間隔??
• ##log文件被保留的時長,如果超過此時長,將會被清除,無論log中的消息是否被消費過.??
• log.retention.hours=168??

?

? ??2.Consumer主要配置

?

Java代碼?? ##當前消費者的group名稱,需要指定??

• group.id=??
• ##比如consumer消費的消息offset等.??
• zookeeper.connect=hostname1:port,hostname2:port2??
• zookeeper.connection.timeout.ms=6000??
• ##當前consumer的標識,可以設定,也可以有系統生成.??
• conusmer.id=??
• ##每次feth將得到多條消息,此值為總大小??
• fetch.messages.max.bytes=1048576??
• fetch.min.bytes=1??
• fetch.wait.max.ms=100??
• ##當有新的consumer加入到group時,將會reblance,此后將會有partitions的消費端遷移到新??
• ##"Partition?Owner?registry"節點信息,但是有可能此時舊的consumer尚沒有釋放此節點,??
• rebalance.max.retries=4??
• ##注意offset信息并不是每消費一次消息就向zk提交一次,而是現在本地保存(內存),并定期提交??
• true??
• auto.commit.interval.ms=60*1000??

?

? ?3.Producer主要配置

?

Java代碼?? ##對于開發者而言,需要通過broker.list指定當前producer需要關注的broker列表??

• ##如果某個broker鏈接失敗,將導致此上的partitons無法繼續發布消息??
• ##對于producer而言沒有使用zookeeper自動發現broker列表，非常奇怪。(0.8V和0.7有區別)??
• ##producer接收消息ack的時機.默認為0.??
• ##1:?當leader接收到消息之后發送ack??
• request.required.acks=0??
• ##如果超時,broker將會向producer發送一個error?ACK.意味著上一次消息因為某種??
• request.timeout.ms=10000??
• ##異步意味著消息將會在本地buffer,并適時批量發送??
• producer.type=sync??
• byte[]??
• class=kafka.serializer.DefaultEncoder??
• class=${serializer.class}??
• ##默認為消息的hashcode?%?partitions個數??
• class=kafka.producer.DefaultPartitioner??
• ##消息壓縮算法,none,gzip,snappy??
• ##消息在producer端buffer的條數.僅在producer.type=async下有效??
• ##在async模式下,當message被緩存的時間超過此值后,將會批量發送給broker??
• queue.buffering.max.ms=5000??
• ##無論如何,producer都無法盡快的將消息發送給broker,從而導致消息在producer端大量沉積??
• queue.buffering.max.messages=10000??
• ##阻塞一定時間后,隊列仍然沒有enqueue(producer仍然沒有發送出任何消息)??
• ##-1:?無阻塞超時限制,消息不會被拋棄??
• queue.enqueue.timeout.ms=-1??
• ##因為broker并沒有完整的機制來避免消息重復,所以當網絡異常時(比如ACK丟失)??
• message.send.max.retries=3??
• ##producer需要知道partition?leader的位置,以及當前topic的情況??
• ##(比如topic失效,partition丟失,leader失效等),此外也可以通過此參數來配置額外的刷新機制??
• topic.metadata.refresh.interval.ms=600000??

?

?

? ? broker配置文件請參考: kafka.server.KafkaConfig

? ? consumer配置文件請參考:?kafka.consumer.ConsumerConfig

? ? producer配置文件請參考:?kafka.producer.ProducerConfig

總結

以上是生活随笔為你收集整理的Apache kafka原理与特性(0.8V)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。