1、銀河麒麟服務(wù)器操作系統(tǒng)Kafka 軟件適配手冊(cè)銀河麒麟服務(wù)器操作系統(tǒng) V4 kafka 軟件適配手冊(cè)I目錄 HYPERLINK l _bookmark0 目錄I HYPERLINK l _bookmark1 概述2 HYPERLINK l _bookmark2 系統(tǒng)概述2 HYPERLINK l _bookmark3 環(huán)境概述2 HYPERLINK l _bookmark4 KAFKA 軟件簡(jiǎn)介2 HYPERLINK l _bookmark5 KAFKA 數(shù)據(jù)流2 HYPERLINK l _bookmark6 KAFKA 使用場(chǎng)景3 HYPERLINK l _bookmark7 KAFKA 設(shè)

2、計(jì)原理4 HYPERLINK l _bookmark8 KAFKA 軟件適配6 HYPERLINK l _bookmark9 下載軟件并解壓6 HYPERLINK l _bookmark10 配置 KAFKA6 HYPERLINK l _bookmark11 啟動(dòng) KAFKA6 HYPERLINK l _bookmark12 KAFKA 測(cè)試6 HYPERLINK l _bookmark13 創(chuàng)建 KAFKA TOPIC6 HYPERLINK l _bookmark14 單 BROKER 測(cè)試7 PAGE 7概述系統(tǒng)概述銀河麒麟服務(wù)器操作系統(tǒng)主要面向軍隊(duì)綜合電子信息系統(tǒng)、金融系統(tǒng)以及電力系統(tǒng)等

3、國家關(guān)鍵行業(yè)的服務(wù)器應(yīng)用領(lǐng)域，突出高安全性、高可用性、高效數(shù)據(jù)處理、虛擬化等關(guān)鍵技術(shù)優(yōu)勢(shì)，針對(duì)關(guān)鍵業(yè)務(wù)構(gòu)建的豐富高效、安全可靠的功能特性，兼容適配長城、聯(lián)想、浪潮、華為、曙光等國內(nèi)主流廠商的服務(wù)器整機(jī)產(chǎn)品，以及達(dá)夢(mèng)、金倉、神通等主要國產(chǎn)數(shù)據(jù)庫和中創(chuàng)、金蝶、東方通等國產(chǎn)中間件，滿足虛擬化、云計(jì)算和大數(shù)據(jù)時(shí)代，服務(wù)器業(yè)務(wù)對(duì)操作系統(tǒng)在性能、安全性及可擴(kuò)展性等方面的需求，是一款具有高安全、高可用、高可靠、高性能的自主可控服務(wù)器操作系統(tǒng)。環(huán)境概述服務(wù)器型號(hào)長城信安擎天 DF720 服務(wù)器CPU 類型飛騰 2000+處理器操作系統(tǒng)版本Kylin-4.0.2-server-sp2-2000-19050910

4、.Z1內(nèi)核版本4.4.131kafka 版本2.11Kafka 軟件簡(jiǎn)介kafka 是一個(gè)分布式的消息隊(duì)列。它提供了類似于 JMS 的特性，但是在設(shè)計(jì)實(shí)現(xiàn)上完全不同，此外它并不是 JMS 規(guī)范的實(shí)現(xiàn)。kafka 對(duì)消息保存時(shí)根據(jù) Topic 進(jìn)行歸類，發(fā)送消息者成為 Producer,消息接受者成為 Consumer,此外 kafka 集群有多個(gè) kafka 實(shí)例組成，每個(gè)實(shí)例(server)成為 broker。無論是 kafka 集群，還是producer 和 consumer 都依賴于 zookeeper 來保證系統(tǒng)可用性集群保存一些 meta 信息。kafka 對(duì)外使用 topic 的概

5、念，生產(chǎn)者往 topic 里寫消息，消費(fèi)者從讀消息。為了做到水平擴(kuò)展，一個(gè) topic 實(shí)際是由多個(gè) partition 組成的，遇到瓶頸時(shí)，可以通過增加 partition 的數(shù)量來進(jìn)行橫向擴(kuò)容。單個(gè) parition 內(nèi)是保證消息有序。每新寫一條消息，kafka 就是在對(duì)應(yīng)的文件 append 寫，所以性能非常高。Kafka 數(shù)據(jù)流kafka 總體數(shù)據(jù)流如下：Producers 往 Brokers 里面的指定 Topic 中寫消息，Consumers 從 Brokers 里面拉去指定 Topic 的消息，然后進(jìn)行業(yè)務(wù)處理。圖中有兩個(gè) topic，topic 0 有兩個(gè)partition，t

6、opic 1 有一個(gè) partition，三副本備份。可以看到 consumer gourp 1 中的consumer 2 沒有分到 partition 處理，這是有可能出現(xiàn)的。關(guān)于 broker、topics、partitions 的一些元信息用 zk 來存，監(jiān)控和路由啥的也都會(huì)用到 zk。Kafka 使用場(chǎng)景消息隊(duì)列對(duì)于一些常規(guī)的消息系統(tǒng)，kafka 是個(gè)不錯(cuò)的選擇；partitons/replication 和容錯(cuò)，可以使 kafka 具有良好的擴(kuò)展性和性能優(yōu)勢(shì)。不過到目前為止，我們應(yīng)該很清楚認(rèn)識(shí)到，kafka 并沒有提供 JMS 中的事務(wù)性消息傳輸擔(dān)保(消息確認(rèn)機(jī)制)消息分組等企業(yè)級(jí)特

7、性；kafka 只能使用作為常規(guī)的消息系統(tǒng)，在一定程度上尚未確保消息的發(fā)送與接收絕對(duì)可靠(比如消息重發(fā)、消息發(fā)送丟失等)。網(wǎng)站活動(dòng)跟蹤kafka 可以作為網(wǎng)站活性跟蹤的最佳工具；可以將網(wǎng)頁/用戶操作等信息發(fā)送到 kafka 中，并實(shí)時(shí)監(jiān)控或者進(jìn)行離線統(tǒng)計(jì)分析。日志聚合kafka 的特性決定它非常適合作為日志收集中心；application 可以將操作日志批量異步的發(fā)送到 kafka 集群中，而不是保存在本地或者 DB 中；kafka 可以批量提交消息/壓縮消息等，這對(duì) producer 端而言，幾乎感覺不到性能的開支。此時(shí) consumer 端可以使 hadoop 等其他系統(tǒng)化的存儲(chǔ)和分析系統(tǒng)

8、。Kafka 設(shè)計(jì)原理kafka 的設(shè)計(jì)初衷是希望作為一個(gè)統(tǒng)一的信息收集平臺(tái)，能夠?qū)崟r(shí)的收集反饋信息，并需要能夠支撐較大的數(shù)據(jù)量,且具備良好的容錯(cuò)能力。持久性kafka 使用文件存儲(chǔ)消息，這就直接決定 kafka 在性能上嚴(yán)重依賴文件系統(tǒng)的本身特性。且無論任何 OS 下，對(duì)文件系統(tǒng)本身的優(yōu)化幾乎沒有可能。文件緩存/ 直接內(nèi)存映射等是常用的手段。因?yàn)?kafka 是對(duì)日志文件進(jìn)行 append 操作，因此磁盤檢索的開支是較小的；同時(shí)為了減少磁盤寫入的次數(shù)，broker 會(huì)將消息暫時(shí)buffer 起來，當(dāng)消息的個(gè)數(shù)(或尺寸)達(dá)到一定閥值時(shí)，再 flush 到磁盤。這樣減少了磁盤 IO 調(diào)用的次數(shù)。

9、性能需要考慮的影響性能點(diǎn)很多，除磁盤 IO 之外，我們還需要考慮網(wǎng)絡(luò) IO，這直接關(guān)系到 kafka 的吞吐量問題。kafka 并沒有提供太多高超的技巧；對(duì)于 producer 端，可以將消息 buffer 起來，當(dāng)消息的條數(shù)達(dá)到一定閥值時(shí)，批量發(fā)送給 broker； 對(duì)于 consumer 端也是一樣，批量 fetch 多條消息。不過消息量的大小可以通過配置文件來指定。對(duì)于 kafka broker 端，似乎有個(gè) sendfile 系統(tǒng)調(diào)用可以潛在的提升網(wǎng)絡(luò) IO 的性能：將文件的數(shù)據(jù)映射到系統(tǒng)內(nèi)存中，socket 直接讀取相應(yīng)的內(nèi)存區(qū)域即可，而無需進(jìn)程再次 copy 和交換。其實(shí)對(duì)于 pr

10、oducer/consumer/broker 三者而言，CPU 的開支應(yīng)該都不大，因此啟用消息壓縮機(jī)制是一個(gè)良好的策略；壓縮需要消耗少量的 CPU 資源，不過對(duì)于 kafka 而言，網(wǎng)絡(luò) IO 更應(yīng)該需要考慮。可以將任何在網(wǎng)絡(luò)上傳輸?shù)南⒍冀?jīng)過壓縮 kafka 支持 gzip/snappy 等多種壓縮方式。生產(chǎn)者負(fù)載均衡: producer 將會(huì)和 Topic 下所有 partition leader 保持 socket 連接；消息由 producer 直接通過 socket 發(fā)送到 broker；中間不會(huì)經(jīng)過任何路由層。事實(shí)上, 消息被路由到哪個(gè) partition 上， 有 produce

11、r 客戶端決定。比如可以采用randomkey-hash輪詢等，如果一個(gè) topic 中有多個(gè) partitions，那么在 producer 端實(shí)現(xiàn)消息均衡分發(fā)是必要的。其中 partition leader 的位置(host:port)注冊(cè)在zookeeper 中，producer 作為 zookeeper client，已經(jīng)注冊(cè)了 watch 用來監(jiān)聽 partitionleader 的變更事件。異步發(fā)送： 將多條消息暫且在客戶端 buffer 起來， 并將他們批量的發(fā)送到broker，小數(shù)據(jù) IO 太多，會(huì)拖慢整體的網(wǎng)絡(luò)延遲，批量延遲發(fā)送事實(shí)上提升了網(wǎng)絡(luò)效率。不過這也有一定的隱患，比如

12、說當(dāng) producer 失效時(shí)，那些尚未發(fā)送的消息將會(huì)丟失。消費(fèi)者consumer 端向 broker 發(fā)送fetch請(qǐng)求， 并告知其獲取消息的 offset；此后consumer 將會(huì)獲得一定條數(shù)的消息；consumer 端也可以重置 offset 來重新消費(fèi)消息。在 JMS 實(shí)現(xiàn)中，Topic 模型基于 push 方式，即 broker 將消息推送給 consumer 端。不過在 kafka 中，采用了 pull 方式，即 consumer 在和 broker 建立連接之后， 主動(dòng)去 pull(或者說 fetch)消息；這中模式有些優(yōu)點(diǎn)，首先 consumer 端可以根據(jù)自己的消費(fèi)能力適時(shí)

13、的去 fetch 消息并處理，且可以控制消息消費(fèi)的進(jìn)度(offset)； 此外，消費(fèi)者可以良好的控制消息消費(fèi)的數(shù)量。其他 JMS 實(shí)現(xiàn)，消息消費(fèi)的位置是有 prodiver 保留，以便避免重復(fù)發(fā)送消息或者將沒有消費(fèi)成功的消息重發(fā)等， 同時(shí)還要控制消息的狀態(tài)。這就要求 JMSbroker 需要太多額外的工作在 kafka 中，partition 中的消息只有一個(gè) consumer 在消費(fèi)，且不存在消息狀態(tài)的控制，也沒有復(fù)雜的消息確認(rèn)機(jī)制，可見 kafka broker 端是相當(dāng)輕量級(jí)的。當(dāng)消息被 consumer 接收之后，consumer 可以在本地保存最后消息的 offset，并間歇性的向

14、zookeeper 注冊(cè) offset。由此可見，consumer 客戶端也很輕量級(jí)。復(fù)制備份kafka 將每個(gè) partition 數(shù)據(jù)復(fù)制到多個(gè) server 上，任何一個(gè) partition 有一個(gè)leader 和多個(gè) follower(可以沒有)；備份的個(gè)數(shù)可以通過 broker 配置文件來設(shè)定。leader 處理所有的 read-write 請(qǐng)求，follower 需要和 leader 保持同步。Follower 和consumer 一樣，消費(fèi)消息并保存在本地日志中；leader 負(fù)責(zé)跟蹤所有的 follower 狀態(tài)，如果 follower落后太多或者失效，leader 將會(huì)把它從

15、 replicas 同步列表中刪除。當(dāng)所有的 follower 都將一條消息保存成功,此消息才被認(rèn)為是committed， 那么此時(shí) consumer 才能消費(fèi)它。即使只有一個(gè) replicas 實(shí)例存活，仍然可以保證消息的正常發(fā)送和接收，只要 zookeeper 集群存活即可。日志如果一個(gè) topic 的名稱為my_topic，它有 2 個(gè) partitions，那么日志將會(huì)保存在 my_topic_0 和 my_topic_1 兩 個(gè) 目 錄 中 ； 日 志 文 件 中 保 存 了 一 序 列 logentries(日志條目)。每個(gè) log entry 格式為4 個(gè)字節(jié)的數(shù)字 N 表示消息

16、的長度 + N 個(gè)字節(jié)的消息內(nèi)容；每個(gè)日志都有一個(gè) offset 來唯一的標(biāo)記一條消息，offset 的值為 8 個(gè)字節(jié)的數(shù)字，表示此消息在此 partition 中所處的起始位置。每個(gè) partition 在物理存儲(chǔ)層面，有多個(gè) log file 組成(稱為 segment).segmentfile 的命名為最小offset，其中每個(gè) partiton 中所持有的 segments 列表信息會(huì)存儲(chǔ)在 zookeeper 中。當(dāng) segment 文件尺寸達(dá)到一定閥值時(shí)(可以通過配置文件設(shè)定,默認(rèn) 1G)，將會(huì)創(chuàng)建一個(gè)新的文件；當(dāng) buffer 中消息的條數(shù)達(dá)到閥值時(shí)將會(huì)觸發(fā)日志信息 flush

17、到日志文件中，同時(shí)如果距離最近一次 flush 的時(shí)間差達(dá)到閥值時(shí)，也會(huì)觸發(fā)flush 到日志文件。如果 broker 失效，極有可能會(huì)丟失那些尚未 flush 到文件的消息。Kafka 軟件適配$ wget HYPERLINK /dist/kafka/kafka_2.11-.tgz /dist/kafka/kafka_2.11 -.tgz$ tar -zxvf kafka_2.11-.tgz -C /usr/local/下載軟件并解壓$ cd /usr/local/kafka_2.11-/$ vim config/perties配置 kafka修改如下：broker.id=0 =localh

18、ostlog.dirs=/usr/local/kafka_2.11-/logsmessage.max.byte=5242880 default.replication.factor=1 replica.fetch.max.bytes=5242880zookeeper.connect=localhost:2181$ bin/kafka-server-start.sh -daemon config/perties啟動(dòng) kafkaKafka 測(cè)試創(chuàng)建 kafka topic$ bin/kafka-topics.sh -zookeeper localhost:2181 -create -topic test -partitions 3-replication-factor 1創(chuàng)建 topic：$ bin/kafka-topics.sh -zookeeper localhost:2181 -list查看所創(chuàng)建的 topic：查看更詳細(xì)的信息：$ bin/kafka-topics.sh -zooke