05.kafka基礎(chǔ)知識(shí)和集群搭建

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

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

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

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

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

6、pertiesbroker.id為依次增長(zhǎng)的：0、1、2、3、4，集群中唯一idlog.dirs設(shè)置到大硬盤(pán)路徑下 work.threadsnum.partitions ，默認(rèn)分區(qū)數(shù)num.io.threads 建議值為機(jī)器的核數(shù)；zookeeper.connect 設(shè)置為zookeeper Servers 列表，各節(jié)點(diǎn)以逗號(hào)分開(kāi)；Kafka 啟動(dòng)、創(chuàng)建topic在kafka的部署目錄下，在各個(gè)節(jié)點(diǎn)上通過(guò)如下命令來(lái)啟動(dòng)：$ 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明細(xì)：./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)和消費(fèi)異常處理報(bào)異常：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 包復(fù)制到kafka libs目錄下面broker.id整數(shù)，

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