本文轉(zhuǎn)載自：http://www.cnblogs.com/likehua/p/3999538.html日熬，作者做了一理解上的的修改

1.什么是kafka?

1.1入門

1.1.1簡介

• kafka是LinkedIn開源的一個分布式MQ系統(tǒng)铸豁，現(xiàn)在是Apache的孵化項目畔塔。

• Kafka is a distributed,partitioned,replicated commit logservice 帅霜，在它的主頁描述kafka為一個高吞吐量的分布式（能將消息分散到不同的節(jié)點上）MQ塞栅。

• Kafka僅僅由7000行Scala編寫卡啰，據(jù)了解岛心，Kafka每秒可以生產(chǎn)約25萬消息（50 MB）役拴，每秒處理55萬消息（110 MB）刑然。

• 提供了類似于JMS的特性，但是在設(shè)計實現(xiàn)上完全不同球涛，此外它并不是JMS的規(guī)范實現(xiàn)。

kafka對消息保存時根據(jù)Topic進(jìn)行歸類校镐，發(fā)送消息者成為Producer,消息接受者成為Consumer,此外kafka集群有多個kafka實例組成亿扁，每個實例(server)成為broker

• kafka目前支持多種客戶端語言：java，python鸟廓，c++从祝，php等等。

• kafka集群的簡要圖解如下肝箱，producer寫入消息哄褒，consumer讀取消息：

image.png

1.1.2 kafka名詞解釋和工作方式：

• Producer ：消息生產(chǎn)者，就是向kafka broker發(fā)消息的客戶端煌张。

• Consumer ：消息消費者呐赡，向kafka broker取消息的客戶端

• Topic ：可以理解為一個隊列。

• Consumer Group （CG）：這是kafka用來實現(xiàn)一個topic消息的廣播（發(fā)給所有的consumer）和單播（發(fā)給任意一個consumer）的手段骏融。一個topic可以有多個CG链嘀。topic的消息會復(fù)制（不是真的復(fù)制萌狂，是概念上的）到所有的CG，但每個CG只會把消息發(fā)給該CG中的一個consumer怀泊。

如果需要實現(xiàn)廣播茫藏，只要每個consumer有一個獨立的CG就可以；

要實現(xiàn)單播只要所有的consumer在同一個CG霹琼；

用CG還可以將consumer進(jìn)行自由的分組而不需要多次發(fā)送消息到不同的topic务傲。

• Broker ：一臺kafka服務(wù)器就是一個broker。一個集群由多個broker組成枣申，一個broker可以容納多個topic

• Partition：為了實現(xiàn)擴(kuò)展性售葡，一個非常大的topic可以分布到多個broker（即服務(wù)器）上，一個topic可以分為多個partition忠藤，每個partition是一個有序的隊列挟伙。partition中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序?qū)⑾l(fā)給consumer模孩，不保證一個topic的整體（多個partition間）的順序尖阔。

• Offset：kafka的存儲文件都是按照offset.kafka來命名，用offset做名字的好處是方便查找榨咐。例如你想找位于2049的位置介却，只要找到2048.kafka的文件即可。當(dāng)然the first offset就是00000000000.kafka

1.2块茁、Topics/logs

一個Topic可以認(rèn)為是一類消息筷笨，每個topic將被分成多個partition(區(qū)),每個partition在存儲層面是append log文件。任何發(fā)布到此partition的消息都會被直接追加到log文件的尾部龟劲，每條消息在文件中的位置稱為offset（偏移量）胃夏，offset為一個long型數(shù)字，它是唯一標(biāo)記一條消息昌跌。它唯一的標(biāo)記一條消息仰禀。kafka并沒有提供其他額外的索引機(jī)制來存儲offset，因為在kafka中幾乎不允許對消息進(jìn)行“隨機(jī)讀寫”蚕愤。

image.png

• kafka和JMS（Java Message Service）實現(xiàn)(activeMQ)不同的是:即使消息被消費,消息仍然不會被立即刪除.日志文件將會根據(jù)broker中的配置要求,保留一定的時間之后刪除;比如log文件保留2天,那么兩天后,文件會被清除,無論其中的消息是否被消費.kafka通過這種簡單的手段,來釋放磁盤空間,以及減少消息消費之后對文件內(nèi)容改動的磁盤IO開支.

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

• kafka集群幾乎不需要維護(hù)任何consumer和producer狀態(tài)信息,這些信息有zookeeper保存;因此producer和consumer的客戶端實現(xiàn)非常輕量級,它們可以隨意離開,而不會對集群造成額外的影響.

• partitions設(shè)計的目的的根本原因

  partitions的設(shè)計目的有多個.最根本原因是kafka基于文件存儲.通過分區(qū),可以將日志內(nèi)容分散到多個server上,來避免文件尺寸達(dá)到單機(jī)磁盤的上限,每個partiton都會被當(dāng)前server(kafka實例)保存;可以將一個topic切分多任意多個partitions,來消息保存/消費的效率.此外越多的partitions意味著可以容納更多的consumer,有效提升并發(fā)消費的能力.(具體原理參見下文).

1.3答恶、Distribution

一個Topic的多個partitions,被分布在kafka集群中的多個server上;每個server(kafka實例)負(fù)責(zé)partitions中消息的讀寫操作;此外kafka還可以配置partitions需要備份的個數(shù)(replicas),每個partition將會被備份到多臺機(jī)器上,以提高可用性.

基于replicated方案,那么就意味著需要對多個備份進(jìn)行調(diào)度;每個partition都有一個server為"leader";leader負(fù)責(zé)所有的讀寫操作,如果leader失效,那么將會有其他follower來接管(成為新的leader);follower只是單調(diào)的和leader跟進(jìn),同步消息即可..由此可見作為leader的server承載了全部的請求壓力,因此從集群的整體考慮,有多少個partitions就意味著有多少個"leader",kafka會將"leader"均衡的分散在每個實例上,來確保整體的性能穩(wěn)定.

• Producers

  Producer將消息發(fā)布到指定的Topic中,同時Producer也能決定將此消息歸屬于哪個partition;比如基于"round-robin"方式或者通過其他的一些算法等.

• Consumers

  本質(zhì)上kafka只支持Topic.每個consumer屬于一個consumer group;反過來說,每個group中可以有多個consumer.發(fā)送到Topic的消息,只會被訂閱此Topic的每個group中的一個consumer消費.

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

  如果所有的consumer都具有不同的group,那這就是"發(fā)布-訂閱";消息將會廣播給所有的消費者.

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

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

• Guarantees

  1. 發(fā)送到partitions中的消息將會按照它接收的順序追加到日志中

  2. 對于消費者而言,它們消費消息的順序和日志中消息順序一致.

  3. 如果Topic的"replicationfactor"為N,那么允許N-1個kafka實例失效.

2、使用場景

2.1 Messaging

對于一些常規(guī)的消息系統(tǒng),kafka是個不錯的選擇;partitons/replication和容錯,可以使kafka具有良好的擴(kuò)展性和性能優(yōu)勢.不過到目前為止,我們應(yīng)該很清楚認(rèn)識到,kafka并沒有提供JMS中的"事務(wù)性""消息傳輸擔(dān)保(消息確認(rèn)機(jī)制)""消息分組"等企業(yè)級特性;kafka只能使用作為"常規(guī)"的消息系統(tǒng),在一定程度上,尚未確保消息的發(fā)送與接收絕對可靠(比如,消息重發(fā),消息發(fā)送丟失等)

2.2 Websit activity tracking

kafka可以作為"網(wǎng)站活性跟蹤"的最佳工具;可以將網(wǎng)頁/用戶操作等信息發(fā)送到kafka中.并實時監(jiān)控,或者離線統(tǒng)計分析等

2.3 Log Aggregation

kafka的特性決定它非常適合作為"日志收集中心";application可以將操作日志"批量""異步"的發(fā)送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/壓縮消息等,這對producer端而言,幾乎感覺不到性能的開支.此時consumer端可以使hadoop等其他系統(tǒng)化的存儲和分析系統(tǒng).

3.設(shè)計原理

kafka的設(shè)計初衷是希望作為一個統(tǒng)一的信息收集平臺,能夠?qū)崟r的收集反饋信息,并需要能夠支撐較大的數(shù)據(jù)量,且具備良好的容錯能力.

3.1萍诱、持久性

kafka使用文件存儲消息,這就直接決定kafka在性能上嚴(yán)重依賴文件系統(tǒng)的本身特性.且無論任何OS下,對文件系統(tǒng)本身的優(yōu)化幾乎沒有可能.文件緩存/直接內(nèi)存映射等是常用的手段.因為kafka是對日志文件進(jìn)行append操作,因此磁盤檢索的開支是較小的;同時為了減少磁盤寫入的次數(shù),broker會將消息暫時buffer起來,當(dāng)消息的個數(shù)(或尺寸)達(dá)到一定閥值時,再flush到磁盤,這樣減少了磁盤IO調(diào)用的次數(shù).

3.2悬嗓、性能

需要考慮的影響性能點很多,除磁盤IO之外,我們還需要考慮網(wǎng)絡(luò)IO,這直接關(guān)系到kafka的吞吐量問題.kafka并沒有提供太多高超的技巧;對于producer端,可以將消息buffer起來,當(dāng)消息的條數(shù)達(dá)到一定閥值時,批量發(fā)送給broker;對于consumer端也是一樣,批量fetch多條消息.不過消息量的大小可以通過配置文件來指定.對于kafka broker端,似乎有個sendfile系統(tǒng)調(diào)用可以潛在的提升網(wǎng)絡(luò)IO的性能:將文件的數(shù)據(jù)映射到系統(tǒng)內(nèi)存中,socket直接讀取相應(yīng)的內(nèi)存區(qū)域即可,而無需進(jìn)程再次copy和交換. 其實對于producer/consumer/broker三者而言,CPU的開支應(yīng)該都不大,因此啟用消息壓縮機(jī)制是一個良好的策略;壓縮需要消耗少量的CPU資源,不過對于kafka而言,網(wǎng)絡(luò)IO更應(yīng)該需要考慮.可以將任何在網(wǎng)絡(luò)上傳輸?shù)南⒍冀?jīng)過壓縮.kafka支持gzip/snappy等多種壓縮方式.

3.3、生產(chǎn)者

負(fù)載均衡: producer將會和Topic下所有partition leader保持socket連接;消息由producer直接通過socket發(fā)送到broker,中間不會經(jīng)過任何"路由層".事實上,消息被路由到哪個partition上,有producer客戶端決定.比如可以采用"random""key-hash""輪詢"等,如果一個topic中有多個partitions,那么在producer端實現(xiàn)"消息均衡分發(fā)"是必要的.

其中partition leader的位置(host:port)注冊在zookeeper中,producer作為zookeeper client,已經(jīng)注冊了watch用來監(jiān)聽partition leader的變更事件.

異步發(fā)送：將多條消息暫且在客戶端buffer起來裕坊，并將他們批量的發(fā)送到broker包竹，小數(shù)據(jù)IO太多，會拖慢整體的網(wǎng)絡(luò)延遲，批量延遲發(fā)送事實上提升了網(wǎng)絡(luò)效率周瞎。不過這也有一定的隱患苗缩，比如說當(dāng)producer失效時，那些尚未發(fā)送的消息將會丟失声诸。

3.4酱讶、消費者

consumer端向broker發(fā)送"fetch"請求,并告知其獲取消息的offset;此后consumer將會獲得一定條數(shù)的消息;consumer端也可以重置offset來重新消費消息.

在JMS實現(xiàn)中,Topic模型基于push方式,即broker將消息推送給consumer端.不過在kafka中,采用了pull方式,即consumer在和broker建立連接之后,主動去pull(或者說fetch)消息;這中模式有些優(yōu)點,首先consumer端可以根據(jù)自己的消費能力適時的去fetch消息并處理,且可以控制消息消費的進(jìn)度(offset);此外,消費者可以良好的控制消息消費的數(shù)量,batch fetch.

其他JMS實現(xiàn),消息消費的位置是有prodiver保留,以便避免重復(fù)發(fā)送消息或者將沒有消費成功的消息重發(fā)等,同時還要控制消息的狀態(tài).這就要求JMS broker需要太多額外的工作.在kafka中,partition中的消息只有一個consumer在消費,且不存在消息狀態(tài)的控制,也沒有復(fù)雜的消息確認(rèn)機(jī)制,可見kafka broker端是相當(dāng)輕量級的.當(dāng)消息被consumer接收之后,consumer可以在本地保存最后消息的offset,并間歇性的向zookeeper注冊offset.由此可見,consumer客戶端也很輕量級.

image.png

3.5、消息傳送機(jī)制

對于JMS實現(xiàn),消息傳輸擔(dān)保非常直接:有且只有一次(exactly once).在kafka中稍有不同:

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

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

3. exactly once: 消息只會發(fā)送一次.

at most once: 消費者fetch消息,然后保存offset,然后處理消息;當(dāng)client保存offset之后,但是在消息處理過程中出現(xiàn)了異常,導(dǎo)致部分消息未能繼續(xù)處理.那么此后"未處理"的消息將不能被fetch到,這就是"at most once".

at least once: 消費者fetch消息,然后處理消息,然后保存offset.如果消息處理成功之后,但是在保存offset階段zookeeper異常導(dǎo)致保存操作未能執(zhí)行成功,這就導(dǎo)致接下來再次fetch時可能獲得上次已經(jīng)處理過的消息,這就是"at least once"彼乌，原因offset沒有及時的提交給zookeeper泻肯，zookeeper恢復(fù)正常還是之前offset狀態(tài).

exactly once: kafka中并沒有嚴(yán)格的去實現(xiàn)(基于2階段提交,事務(wù)),我們認(rèn)為這種策略在kafka中是沒有必要的.

通常情況下"at-least-once"是我們搜選.(相比at most once而言,重復(fù)接收數(shù)據(jù)總比丟失數(shù)據(jù)要好).

3.6、復(fù)制備份

kafka將每個partition數(shù)據(jù)復(fù)制到多個server上,任何一個partition有一個leader和多個follower(可以沒有);備份的個數(shù)可以通過broker配置文件來設(shè)定.leader處理所有的read-write請求,follower需要和leader保持同步.Follower和consumer一樣,消費消息并保存在本地日志中;leader負(fù)責(zé)跟蹤所有的follower狀態(tài),如果follower"落后"太多或者失效,leader將會把它從replicas同步列表中刪除.當(dāng)所有的follower都將一條消息保存成功,此消息才被認(rèn)為是"committed",那么此時consumer才能消費它.即使只有一個replicas實例存活,仍然可以保證消息的正常發(fā)送和接收,只要zookeeper集群存活即可.(不同于其他分布式存儲,比如hbase需要"多數(shù)派"存活才行)

當(dāng)leader失效時,需在followers中選取出新的leader,可能此時follower落后于leader,因此需要選擇一個"up-to-date"的follower.選擇follower時需要兼顧一個問題,就是新leaderserver上所已經(jīng)承載的partition leader的個數(shù),如果一個server上有過多的partition leader,意味著此server將承受著更多的IO壓力.在選舉新leader,需要考慮到"負(fù)載均衡".

3.7.日志

如果一個topic的名稱為"my_topic",它有2個partitions,那么日志將會保存在my_topic_0和my_topic_1兩個目錄中;日志文件中保存了一序列"log entries"(日志條目),每個log entry格式為"4個字節(jié)的數(shù)字N表示消息的長度" + "N個字節(jié)的消息內(nèi)容";每個日志都有一個offset來唯一的標(biāo)記一條消息,offset的值為8個字節(jié)的數(shù)字,表示此消息在此partition中所處的起始位置..每個partition在物理存儲層面,有多個log file組成(稱為segment).segmentfile的命名為"最小offset".kafka.例如"00000000000.kafka";其中"最小offset"表示此segment中起始消息的offset.

image.png

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

當(dāng)segment文件尺寸達(dá)到一定閥值時(可以通過配置文件設(shè)定,默認(rèn)1G),將會創(chuàng)建一個新的文件;當(dāng)buffer中消息的條數(shù)達(dá)到閥值時將會觸發(fā)日志信息flush到日志文件中,同時如果"距離最近一次flush的時間差"達(dá)到閥值時,也會觸發(fā)flush到日志文件.如果broker失效,極有可能會丟失那些尚未flush到文件的消息.因為server意外實現(xiàn),仍然會導(dǎo)致log文件格式的破壞(文件尾部),那么就要求當(dāng)server啟東是需要檢測最后一個segment的文件結(jié)構(gòu)是否合法并進(jìn)行必要的修復(fù).

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

日志文件的刪除策略非常簡單:啟動一個后臺線程定期掃描log file列表,把保存時間超過閥值的文件直接刪除(根據(jù)文件的創(chuàng)建時間).為了避免刪除文件時仍然有read操作(consumer消費),采取copy-on-write方式.

3.8慰照、kafka與zookeeper之間的聯(lián)系

• kafka使用zookeeper來實現(xiàn)動態(tài)的集群擴(kuò)展软免，不需要更改客戶端（producer和consumer）的配置。broker會在zookeeper注冊并保持相關(guān)的元數(shù)據(jù)（topic焚挠，partition信息等）更新。

• 而客戶端會在zookeeper上注冊相關(guān)的watcher漓骚。一旦zookeeper發(fā)生變化蝌衔，客戶端能及時感知并作出相應(yīng)調(diào)整。這樣就保證了添加或去除broker時蝌蹂，各broker間仍能自動實現(xiàn)負(fù)載均衡噩斟。

  kafka使用zookeeper來存儲一些meta信息,并使用了zookeeper watch機(jī)制來發(fā)現(xiàn)meta信息的變更并作出相應(yīng)的動作(比如consumer失效,觸發(fā)負(fù)載均衡等)

  1. Broker node registry: 當(dāng)一個kafkabroker啟動后,首先會向zookeeper注冊自己的節(jié)點信息(臨時znode),同時當(dāng)broker和zookeeper斷開連接時,此znode也會被刪除.

  格式: /broker/ids/[0...N] -->host:port;其中[0..N]表示broker id,每個broker的配置文件中都需要指定一個數(shù)字類型的id(全局不可重復(fù)),znode的值為此broker的host:port信息.

  2. Broker Topic Registry: 當(dāng)一個broker啟動時,會向zookeeper注冊自己持有的topic和partitions信息,仍然是一個臨時znode.

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

  3. Consumer and Consumer group: 每個consumer客戶端被創(chuàng)建時,會向zookeeper注冊自己的信息;此作用主要是為了"負(fù)載均衡".

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

  4. Consumer id Registry: 每個consumer都有一個唯一的ID(host:uuid,可以通過配置文件指定,也可以由系統(tǒng)生成),此id用來標(biāo)記消費者信息.

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

  仍然是一個臨時的znode,此節(jié)點的值為{"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為持久節(jié)點,可以看出offset跟group_id有關(guān),以表明當(dāng)group中一個消費者失效,其他consumer可以繼續(xù)消費.

  6. Partition Owner registry: 用來標(biāo)記partition被哪個consumer消費.臨時znode

  格式:/consumers/[group_id]/owners/[topic]/[broker_id-partition_id]-->consumer_node_id當(dāng)consumer啟動時,所觸發(fā)的操作:

  A) 首先進(jìn)行"Consumer id Registry";

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

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

image.png

1. Producer端使用zookeeper用來"發(fā)現(xiàn)"broker列表,以及和Topic下每個partition leader建立socket連接并發(fā)送消息.

2. Broker端使用zookeeper用來注冊broker信息,已經(jīng)監(jiān)測partitionleader存活性.

3. Consumer端使用zookeeper用來注冊consumer信息,其中包括consumer消費的partition列表等,同時也用來發(fā)現(xiàn)broker列表,并和partition leader建立socket連接,并獲取消息.

4、主要配置

4.1孤个、Broker配置

image.png

4.2.Consumer主要配置

image.png

4.3.Producer主要配置

image.png

以上是關(guān)于kafka一些基礎(chǔ)說明剃允，在其中我們知道如果要kafka正常運(yùn)行，必須配置zookeeper齐鲤，否則無論是kafka集群還是客戶端的生存者和消費者都無法正常的工作的

歡迎大家掃碼關(guān)注我的微信公眾號斥废，與大家一起分享技術(shù)與成長中的故事。

我的微信公眾號.jpg