1、Storm企業(yè)項目實戰(zhàn)（含多個完整應用）講師：Cloudy（北風網(wǎng) )答疑 5、Kafka基礎(chǔ)和集群搭建Kafka課程采用：Kafka是LinkedIn開源出來的一個高吞吐的分布式消息系統(tǒng)。使用scala開發(fā)，支持多語言客戶端（c+、java、python等）其具有以下特點：1) 支持高Throughput（高吞吐量）的應用2) 無需停機即可擴展機器3) 持久化：通過將數(shù)據(jù)持久化到硬盤以及replication防止數(shù)據(jù)丟失4) 支持online和offline的場景。Kafka架構(gòu)Broker：消息中間件處理結(jié)點，一個kafka節(jié)點就是一個brokerProducer：消息發(fā)布者Consume

2、r：消息訂閱者消息kafka的消息分幾個層次：1) Topic：一類消息，例如page view日志，click日志等都可以以topic的形式存在，kafka集群能夠同時負責多個topic的分發(fā)2) Partition： Topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。3) Message：消息，最小訂閱單元消息處理流程數(shù)據(jù)流程：1. Producer根據(jù)指定的partition方法（round-robin、hash等），將消息發(fā)布到指定topic的partit

3、ion里面2. kafka集群接收到Producer發(fā)過來的消息后，將其持久化到硬盤，并保留消息指定時長（可配置），而不關(guān)注消息是否被消費。3. Consumer從kafka集群消費（pull或push方式）數(shù)據(jù)，并控制獲取消息的offset偏移量Kafka如何實現(xiàn)高吞吐量？High Throughput是kafka需要實現(xiàn)的核心目標之一，為此kafka做了以下一些設(shè)計：1）數(shù)據(jù)磁盤持久化：消息不在內(nèi)存中cache，直接寫入到磁盤，充分利用磁盤的順序讀寫性能，所以broker沒有內(nèi)存壓力。2）zero-copy：減少IO操作步驟3）數(shù)據(jù)批量發(fā)送4）數(shù)據(jù)壓縮5）Topic劃分為多個partiti

4、on，提高parallelism （并行度）負載均衡 load balance&HAKafka如何實現(xiàn)負載均衡的？1) producer根據(jù)用戶指定的算法，將消息發(fā)送到指定的partition2) 存在多個partiiton，每個partition有自己的replica（副本），每個replica分布在不同的Broker節(jié)點上3) 多個partition需要選取出lead partition，leader partition負責讀寫，并由zookeeper負責fail over4) 通過zookeeper管理broker與consumer的動態(tài)加入與離開擴容當需要增加broker結(jié)點時，新增的

5、broker會向zookeeper注冊，而producer及consumer會根據(jù)注冊在zookeeper上的watcher感知這些變化，并及時作出調(diào)整。集群搭建步驟1. Kafka下載：wget 解壓 tar zxvf kafka_2.9.2-0.8.1.tgzcd kafka_2.9.2-0.8.1kafka使用scala編寫，需要下載scala相關(guān)的庫下載安裝sbt：wget $ rpm -ivh sbt.rpm集群搭建步驟3. 更新scala環(huán)境：sbt update sbt package #sbt assembly-package-dependency4. 配置config/ser

6、pertiesbroker.id為依次增長的：0、1、2、3、4，集群中唯一idlog.dirs設(shè)置到大硬盤路徑下 work.threadsnum.partitions ，默認分區(qū)數(shù)num.io.threads 建議值為機器的核數(shù)；zookeeper.connect 設(shè)置為zookeeper Servers 列表，各節(jié)點以逗號分開；Kafka 啟動、創(chuàng)建topic在kafka的部署目錄下，在各個節(jié)點上通過如下命令來啟動：$ nohup bin/kafka-server-start.sh config/perties &rootmaster # jps3098

7、ZooKeeperMain3037 Kafka創(chuàng)建topic：$ bin/kafka-topics.sh -zookeeper 07:2181,08:2181 -topic topicName -replication-factor 1 -partitions 1 -create其中， -topic 定義topic名 -replication-factor 定義副本數(shù) -partitions 定義分區(qū)數(shù)查看全部Topic：bin/kafka-topics.sh -zookeeper 07:2181,07:2

8、181 -list刪除Topic：./kafka-topics.sh -topic topicName -delete -zookeeper 07:2181,07:2181查看Topic明細：./kafka-topics.sh -topic test -describe -zookeeper 07:2181,07:2181Productor：bin/kafka-console-producer.sh -broker-list07:9092,08:9092 -top

9、ic topicNameConsumer:bin/kafka-console-consumer.sh -zookeeper 07:2181,07:2181 -topic topicName -from-beginning生產(chǎn)和消費異常處理報異常：Failed to load class org.slf4j.impl.StaticLoggerBinder解決方法：下載slf4j-1.7.6.zipwget 解壓unzip slf4j-1.7.6.zip把slf4j-nop-1.7.6.jar 包復制到kafka libs目錄下面broker.id整數(shù)，

10、建議根據(jù)ip區(qū)分log.dirskafka存放消息文件的路徑，默認/tmp/kafka-logsportbroker用于接收producer消息的端口zookeeper.connnectzookeeper連接格式為 ip1:port,ip2:port,ip3:portmessage.max.bytes單條消息的最大長度 work.threadsbroker用于處理網(wǎng)絡(luò)請求的線程數(shù)如不配置默認為3，perties默認是2num.io.threadsbroker用于執(zhí)行網(wǎng)絡(luò)請求的IO線程數(shù)如不配置默認為8，perties默認是2可適當增大，queued.max.requests排隊等候IO線程執(zhí)行的requests默認為500broker的hostname默認null,建議寫主機的ip,不然消費端不配置hosts會有麻煩num.partitionstopic的默認分區(qū)數(shù)默認2log.retention.hours消息被刪除前保存多少小時默認1周168小時auto.create.topics.enable是