Storm上游數據源之Kakfa
目標：
理解Storm消費的數據來源、理解JMS規范、理解Kafka核心組件、掌握Kakfa生產者API、掌握Kafka消費者API。對流式計算的生態環境有深入的了解，具備流式計算項目架構的能力。
大綱：
1、 kafka是什么？
2、 JMS規范是什么？
3、 為什么需要消息隊列？
4、 Kafka核心組件
5、 Kafka安裝部署
6、 Kafka生產者Java API
7、 Kafka消費者Java API
內容
1、Kafka是什么
在流式計算中，Kafka一般用來緩存數據，Storm通過消費Kafka的數據進行計算。
KAFKA + STORM +REDIS
l Apache Kafka是一個開源消息系統，由Scala寫成。是由Apache軟件基金會開發的一個開源消息系統項目。
l Kafka最初是由LinkedIn開發，並於2011年初開源。2012年10月從Apache Incubator畢業。該項目的目標是為處理實時數據提供一個統一、高通量、低等待的平台。
l Kafka是一個分布式消息隊列：生產者、消費者的功能。它提供了類似於JMS的特性，但是在設計實現上完全不同，此外它並不是JMS規范的實現。
l Kafka對消息保存時根據Topic進行歸類，發送消息者稱為Producer,消息接受者稱為Consumer,此外kafka集群有多個kafka實例組成，每個實例(server)成為broker。
l 無論是kafka集群，還是producer和consumer都依賴於zookeeper集群保存一些meta信息，來保證系統可用性
2、JMS是什么
2.1、JMS的基礎
JMS是什么：JMS是Java提供的一套技術規范
JMS干什么用：用來異構系統 集成通信，緩解系統瓶頸，提高系統的伸縮性增強系統用戶體驗，使得系統模塊化和組件化變得可行並更加靈活
通過什么方式：生產消費者模式（生產者、服務器、消費者）
jdk，kafka，activemq……
2.2、JMS消息傳輸模型
l 點對點模式（一對一，消費者主動拉取數據，消息收到后消息清除）
點對點模型通常是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特點是發送到隊列的消息被一個且只有一個接收者接收處理，即使有多個消息監聽者也是如此。
l 發布/訂閱模式（一對多，數據生產后，推送給所有訂閱者）
發布訂閱模型則是一個基於推送的消息傳送模型。發布訂閱模型可以有多種不同的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的所有消息，即時當前訂閱者不可用，處於離線狀態。
queue.put（object） 數據生產
queue.take(object) 數據消費
2.3、JMS核心組件
l Destination：消息發送的目的地，也就是前面說的Queue和Topic。
l Message [m1] ：從字面上就可以看出是被發送的消息。
l Producer： 消息的生產者，要發送一個消息，必須通過這個生產者來發送。
l MessageConsumer： 與生產者相對應，這是消息的消費者或接收者，通過它來接收一個消息。
通過與ConnectionFactory可以獲得一個connection
通過connection可以獲得一個session會話。
2.4、常見的類JMS消息服務器
2.4.1、JMS消息服務器 ActiveMQ
ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的。
主要特點：
l 多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
l 完全支持JMS1.1和J2EE 1.4規范 (持久化,XA消息,事務)
l 對Spring的支持,ActiveMQ可以很容易內嵌到使用Spring的系統里面去,而且也支持Spring2.0的特性
l 通過了常見J2EE服務器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中通過JCA 1.5 resource adaptors的配置,可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上
l 支持多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
l 支持通過JDBC和journal提供高速的消息持久化
l 從設計上保證了高性能的集群,客戶端-服務器,點對點
l 支持Ajax
l 支持與Axis的整合
l 可以很容易得調用內嵌JMS provider,進行測試
2.4.2、分布式消息中間件 Metamorphosis
Metamorphosis (MetaQ) 是一個高性能、高可用、可擴展的分布式消息中間件，類似於LinkedIn的Kafka，具有消息存儲順序寫、吞吐量大和支持本地和XA事務等特性，適用於大吞吐量、順序消息、廣播和日志數據傳輸等場景，在淘寶和支付寶有着廣泛的應用，現已開源。
主要特點：
l 生產者、服務器和消費者都可分布
l 消息存儲順序寫
l 性能極高,吞吐量大
l 支持消息順序
l 支持本地和XA事務
l 客戶端pull，隨機讀,利用sendfile系統調用，zero-copy ,批量拉數據
l 支持消費端事務
l 支持消息廣播模式
l 支持異步發送消息
l 支持http協議
l 支持消息重試和recover
l 數據遷移、擴容對用戶透明
l 消費狀態保存在客戶端
l 支持同步和異步復制兩種HA
l 支持group commit
2.4.3、分布式消息中間件 RocketMQ
RocketMQ 是一款分布式、隊列模型的消息中間件，具有以下特點：
l 能夠保證嚴格的消息順序
l 提供豐富的消息拉取模式
l 高效的訂閱者水平擴展能力
l 實時的消息訂閱機制
l 億級消息堆積能力
l Metaq3.0 版本改名，產品名稱改為RocketMQ
2.4.4、其他MQ
l .NET消息中間件 DotNetMQ
l 基於HBase的消息隊列 HQueue
l Go 的 MQ 框架 KiteQ
l AMQP消息服務器 RabbitMQ
l MemcacheQ 是一個基於 MemcacheDB 的消息隊列服務器。
3、為什么需要消息隊列（重要）
消息系統的核心作用就是三點：解耦，異步和並行
以用戶注冊的案列來說明消息系統的作用
3.1、用戶注冊的一般流程
問題：隨着后端流程越來越多，每步流程都需要額外的耗費很多時間，從而會導致用戶更長的等待延遲。
3.2、用戶注冊的並行執行
問題：系統並行的發起了4個請求，4個請求中，如果某一個環節執行1分鍾，其他環節再快，用戶也需要等待1分鍾。如果其中一個環節異常之后，整個服務掛掉了。
3.3、用戶注冊的最終一致
1、 保證主流程的正常執行、執行成功之后，發送MQ消息出去。
2、 需要這個destination的其他系統通過消費數據再執行，最終一致。
4、Kafka核心組件
l Topic ：消息根據Topic進行歸類
l Producer：發送消息者
l Consumer：消息接受者
l broker：每個kafka實例(server)
l Zookeeper：依賴集群保存meta信息。
5、Kafka集群部署
5.1集群部署的基本流程
下載安裝包、解壓安裝包、修改配置文件、分發安裝包、啟動集群
5.2集群部署的基礎環境准備
安裝前的准備工作（zk集群已經部署完畢）
l 關閉防火牆
chkconfig iptables off && setenforce 0
l 創建用戶
groupadd realtime && useradd realtime && usermod -a -G realtime realtime
l 創建工作目錄並賦權
mkdir /export
mkdir /export/servers
chmod 755 -R /export
l 切換到realtime用戶下
su realtime
5.3 Kafka集群部署
5.3.1、下載安裝包
http://kafka.apache.org/downloads.html
在linux中使用wget命令下載安裝包
wget http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz
5.3.2、解壓安裝包
tar -zxvf /export/software/kafka_2.11-0.8.2.2.tgz -C /export/servers/
cd /export/servers/
ln -s kafka_2.11-0.8.2.2 kafka
5.3.3、修改配置文件
cp /export/servers/kafka/config/server.properties
/export/servers/kafka/config/server.properties.bak
vi /export/servers/kafka/config/server.properties
輸入以下內容：
5.3.4、分發安裝包
scp -r /export/servers/kafka_2.11-0.8.2.2 kafka02:/export/servers
然后分別在各機器上創建軟連
cd /export/servers/
ln -s kafka_2.11-0.8.2.2 kafka
5.3.5、再次修改配置文件（重要）
依次修改各服務器上配置文件的的broker.id，分別是0,1,2不得重復。
5.3.6、啟動集群
首先啟動zookeeper集群
zkServer.sh start
依次在各節點上啟動kafka
bin/kafka-server-start.sh config/server.properties
集群啟動后會卡主頁面，需要再開一個窗口調用jps指令查看kafka是否已經啟動
5.4、Kafka常用操作命令
l 查看當前服務器中的所有topic
bin/kafka-topics.sh --list --zookeeper zk01:2181
l 創建topic
./kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 1 --partitions 3 --topic first
l 刪除topic
sh bin/kafka-topics.sh --delete --zookeeper zk01:2181 --topic test
需要server.properties中設置delete.topic.enable=true否則只是標記刪除或者直接重啟。
l 通過shell命令發送消息
kafka-console-producer.sh --broker-list kafka01:9092 --topic itheima
l 通過shell消費消息
sh bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic test1
l 查看消費位置
sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zk01:2181 --group testGroup
l 查看某個Topic的詳情
sh kafka-topics.sh --topic test --describe --zookeeper zk01:2181
6、Kafka生產者Java API
7、Kafka消費者Java API
StreamMessage：Java 數據流消息，用標准流操作來順序的填充和讀取。
MapMessage：一個Map類型的消息；名稱為 string 類型，而值為 Java 的基本類型。
TextMessage：普通字符串消息，包含一個String。
ObjectMessage：對象消息，包含一個可序列化的Java 對象
BytesMessage：二進制數組消息，包含一個byte[]。
XMLMessage: 一個XML類型的消息。
最常用的是TextMessage和ObjectMessage。
Kafka負載均衡、Kafka自定義Partition、Kafk文件存儲機制介紹
名稱：
Kafka技術增強
注：請先學習Kafka基礎
目標：
通過本模塊的學習，能夠掌握Kafka的負載均衡、Producer生產數據、Kafka文件存儲機制、Kafka自定義partition
課程大綱：
1、 Kafka整體結構圖
2、 Consumer與topic關系
3、 Kafka Producer消息分發
4、 Consumer 的負載均衡
5、 Kafka文件存儲機制
內容
1、Kafka整體結構圖
Kafka名詞解釋和工作方式
l Producer ：消息生產者，就是向kafka broker發消息的客戶端。
l Consumer ：消息消費者，向kafka broker取消息的客戶端
l Topic ：咱們可以理解為一個隊列。目標發送的目的地，這是一個邏輯上的概念，落到磁盤上是一個partition的目錄。
partition的目錄中有多個segment組合(index,log)；一個Topic對應多個partition[0,1,2,3]，一個partition對應多個segment組合。一個segment有默認的大小是1G。每個partition可以設置多個副本(replication-factor 1),會從所有的副本中選取一個leader出來。所有讀寫操作都是通過leader來進行的。
特別強調，和mysql中主從有區別，mysql做主從是為了讀寫分離，在kafka中讀寫操作都是leader。
l Consumer Group （CG）：這是kafka用來實現一個topic消息的廣播（發給所有的consumer）和單播（發給任意一個consumer）的手段。一個topic可以有多個CG。topic的消息會復制（不是真的復制，是概念上的）到所有的CG，但每個partion只會把消息發給該CG中的一個consumer。如果需要實現廣播，只要每個consumer有一個獨立的CG就可以了。要實現單播只要所有的consumer在同一個CG。用CG還可以將consumer進行自由的分組而不需要多次發送消息到不同的topic。
l Broker ：一台kafka服務器就是一個broker。一個集群由多個broker組成。一個broker可以容納多個topic。
l Partition：為了實現擴展性，一個非常大的topic可以分布到多個broker（即服務器）上，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的整體（多個partition間）的順序。
l Offset：kafka的存儲文件都是按照offset.kafka來命名，用offset做名字的好處是方便查找。例如你想找位於2049的位置，只要找到2048.kafka的文件即可。當然the first offset就是00000000000.kafka
2、Consumer與topic關系
本質上kafka只支持Topic；
l 每個group中可以有多個consumer，每個consumer屬於一個consumer group；
通常情況下，一個group中會包含多個consumer，這樣不僅可以提高topic中消息的並發消費能力，而且還能提高"故障容錯"性，如果group中的某個consumer失效那么其消費的partitions將會有其他consumer自動接管。
l 對於Topic中的一條特定的消息，只會被訂閱此Topic的每個group中的其中一個consumer消費，此消息不會發送給一個group的多個consumer；
那么一個group中所有的consumer將會交錯的消費整個Topic，每個group中consumer消息消費互相獨立，我們可以認為一個group是一個"訂閱"者。
l 在kafka中,一個partition中的消息只會被group中的一個consumer消費(同一時刻)；
一個Topic中的每個partions，只會被一個"訂閱者"中的一個consumer消費，不過一個consumer可以同時消費多個partitions中的消息。
l kafka的設計原理決定,對於一個topic，同一個group中不能有多於partitions個數的consumer同時消費，否則將意味着某些consumer將無法得到消息。
kafka只能保證一個partition中的消息被某個consumer消費時是順序的；事實上，從Topic角度來說,當有多個partitions時,消息仍不是全局有序的。
3、Kafka消息的分發
Producer客戶端負責消息的分發
l kafka集群中的任何一個broker都可以向producer提供metadata信息,這些metadata中包含"集群中存活的servers列表"/"partitions leader列表"等信息；
l 當producer獲取到metadata信息之后, producer將會和Topic下所有partition leader保持socket連接；
l 消息由producer直接通過socket發送到broker，中間不會經過任何"路由層"，事實上，消息被路由到哪個partition上由producer客戶端決定；
比如可以采用"random""key-hash""輪詢"等,如果一個topic中有多個partitions,那么在producer端實現"消息均衡分發"是必要的。
l 在producer端的配置文件中,開發者可以指定partition路由的方式。
Producer消息發送的應答機制
設置發送數據是否需要服務端的反饋,有三個值0,1,-1
0: producer不會等待broker發送ack
1: 當leader接收到消息之后發送ack
-1: 當所有的follower都同步消息成功后發送ack
request.required.acks=0
4、Consumer的負載均衡
當一個group中,有consumer加入或者離開時,會觸發partitions均衡.均衡的最終目的,是提升topic的並發消費能力，步驟如下：
1、 假如topic1,具有如下partitions: P0,P1,P2,P3
2、 加入group中,有如下consumer: C1,C2
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)]
5、kafka文件存儲機制
5.1、Kafka文件存儲基本結構
l 在Kafka文件存儲中，同一個topic下有多個不同partition，每個partition為一個目錄，partiton命名規則為topic名稱+有序序號，第一個partiton序號從0開始，序號最大值為partitions數量減1。
l 每個partion(目錄)相當於一個巨型文件被平均分配到多個大小相等segment(段)數據文件中。但每個段segment file消息數量不一定相等，這種特性方便old segment file快速被刪除。默認保留7天的數據。
l 每個partiton只需要支持順序讀寫就行了，segment文件生命周期由服務端配置參數決定。（什么時候創建，什么時候刪除）
數據有序的討論？
一個partition的數據是否是有序的？ 間隔性有序，不連續
針對一個topic里面的數據，只能做到partition內部有序，不能做到全局有序。
特別加入消費者的場景后，如何保證消費者消費的數據全局有序的？偽命題。
只有一種情況下才能保證全局有序？就是只有一個partition。
5.2、Kafka Partition Segment
l Segment file組成：由2大部分組成，分別為index file和data file，此2個文件一一對應，成對出現，后綴".index"和“.log”分別表示為segment索引文件、數據文件。
l Segment文件命名規則：partion全局的第一個segment從0開始，后續每個segment文件名為上一個segment文件最后一條消息的offset值。數值最大為64位long大小，19位數字字符長度，沒有數字用0填充。
l 索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。
3，497：當前log文件中的第幾條信息，存放在磁盤上的那個地方
上述圖中索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。
其中以索引文件中元數據3,497為例，依次在數據文件中表示第3個message(在全局partiton表示第368772個message)、以及該消息的物理偏移地址為497。
l segment data file由許多message組成， qq物理結構如下：
| 關鍵字 |
解釋說明 |
| 8 byte offset |
在parition(分區)內的每條消息都有一個有序的id號，這個id號被稱為偏移(offset),它可以唯一確定每條消息在parition(分區)內的位置。即offset表示partiion的第多少message |
| 4 byte message size |
message大小 |
| 4 byte CRC32 |
用crc32校驗message |
| 1 byte “magic" |
表示本次發布Kafka服務程序協議版本號 |
| 1 byte “attributes" |
表示為獨立版本、或標識壓縮類型、或編碼類型。 |
| 4 byte key length |
表示key的長度,當key為-1時，K byte key字段不填 |
| K byte key |
可選 |
| value bytes payload |
表示實際消息數據。 |
5.3、Kafka 查找message
讀取offset=368776的message，需要通過下面2個步驟查找。
5.3.1、查找segment file
00000000000000000000.index表示最開始的文件，起始偏移量(offset)為0
00000000000000368769.index的消息量起始偏移量為368770 = 368769 + 1
00000000000000737337.index的起始偏移量為737338=737337 + 1
其他后續文件依次類推。
以起始偏移量命名並排序這些文件，只要根據offset **二分查找**文件列表，就可以快速定位到具體文件。當offset=368776時定位到00000000000000368769.index和對應log文件。
5.3.2、通過segment file查找message
當offset=368776時，依次定位到00000000000000368769.index的元數據物理位置和00000000000000368769.log的物理偏移地址
然后再通過00000000000000368769.log順序查找直到offset=368776為止。
6、Kafka自定義Partition
見代碼