phpkafka訂閱,php kafka 消費

本文目錄一覽：

• 1、kafka——消費者原理解析
• 2、消息中間件Kafka – PHP操作使用Kafka
• 3、大數據Kafka有哪些優勢能力呢？
• 4、大數據Kafka是什麼呢？
• 5、大型的PHP應用，通常使用什麼應用做消息隊列？
• 6、Kafka使用場景

kafka——消費者原理解析

kafka採用發佈訂閱模式：一對多。發佈訂閱模式又分兩種：

Kafka為這兩種模型提供了單一的消費者抽象模型： 消費者組 （consumer group）。 消費者用一個消費者組名標記自己。 一個發佈在Topic上消息被分發給此消費者組中的一個消費者。 假如所有的消費者都在一個組中，那麼這就變成了隊列模型。 假如所有的消費者都在不同的組中，那麼就完全變成了發佈-訂閱模型。 一個消費者組中消費者訂閱同一個Topic，每個消費者接受Topic的一部分分區的消息，從而實現對消費者的橫向擴展，對消息進行分流。

注意：當單個消費者無法跟上數據生成的速度，就可以增加更多的消費者分擔負載，每個消費者只處理部分partition的消息，從而實現單個應用程序的橫向伸縮。但是不要讓消費者的數量多於partition的數量，此時多餘的消費者會空閑。此外，Kafka還允許多個應用程序從同一個Topic讀取所有的消息，此時只要保證每個應用程序有自己的消費者組即可。

消費者組的概念就是：當有多個應用程序都需要從Kafka獲取消息時，讓每個app對應一個消費者組，從而使每個應用程序都能獲取一個或多個Topic的全部消息；在每個消費者組中，往消費者組中添加消費者來伸縮讀取能力和處理能力，消費者組中的每個消費者只處理每個Topic的一部分的消息，每個消費者對應一個線程。

在同一個群組中，無法讓一個線程運行多個消費者，也無法讓多線線程安全地共享一個消費者。按照規則，一個消費者使用一個線程，如果要在同一個消費者組中運行多個消費者，需要讓每個消費者運行在自己的線程中。最好把消費者的邏輯封裝在自己的對象中，然後使用java的ExecutorService啟動多個線程，使每個消費者運行在自己的線程上,可參考

一個 consumer group 中有多個 consumer，一個 topic 有多個 partition，所以必然會涉及到 partition 的分配問題，即確定哪個 partition 由哪個 consumer 來消費。

關於如何設置partition值需要考慮的因素

Kafka 有兩種分配策略，一個是 RoundRobin，一個是 Range，默認為Range，當消費者組內消費者發生變化時，會觸發分區分配策略（方法重新分配）。

以上三種現象會使partition的所有權在消費者之間轉移，這樣的行為叫作再均衡。

再均衡的優點 ：

再均衡的缺點 ：

RoundRobin 輪詢方式將分區所有作為一個整體進行 Hash 排序，消費者組內分配分區個數最大差別為 1，是按照組來分的，可以解決多個消費者消費數據不均衡的問題。

但是，當消費者組內訂閱不同主題時，可能造成消費混亂，如下圖所示，Consumer0 訂閱主題 A，Consumer1 訂閱主題 B。

將 A、B 主題的分區排序後分配給消費者組，TopicB 分區中的數據可能 分配到 Consumer0 中。

Range 方式是按照主題來分的，不會產生輪詢方式的消費混亂問題。

但是，如下圖所示，Consumer0、Consumer1 同時訂閱了主題 A 和 B，可能造成消息分配不對等問題，當消費者組內訂閱的主題越多，分區分配可能越不均衡。

由於 consumer 在消費過程中可能會出現斷電宕機等故障，consumer 恢復後，需要從故障前的位置繼續消費，所以 consumer 需要實時記錄自己消費到了哪個 offset，以便故障恢復後繼續消費。

consumer group +topic + partition 唯一確定一個offest

Kafka 0.9 版本之前，consumer 默認將 offset 保存在 Zookeeper 中，從 0.9 版本開始，

consumer 默認將 offset 保存在 Kafka 一個內置的 topic 中，該 topic 為__consumer_offsets。

你如果特別好奇，實在想看看offset什麼的，也可以執行下面操作：

修改配置文件 consumer.properties

再啟動一個消費者

當消費者崩潰或者有新的消費者加入，那麼就會觸發再均衡（rebalance），完成再均衡後，每個消費者可能會分配到新的分區，而不是之前處理那個，為了能夠繼續之前的工作，消費者需要讀取每個partition最後一次提交的偏移量，然後從偏移量指定的地方繼續處理。

case1：如果提交的偏移量小於客戶端處理的最後一個消息的偏移量，那麼處於兩個偏移量之間的消息就會被重複處理。

case2：如果提交的偏移量大於客戶端處理的最後一個消息的偏移量，那麼處於兩個偏移量之間的消息將會丟失。

自動提交的優點是方便，但是可能會重複處理消息

不足：broker在對提交請求作出回應之前，應用程序會一直阻塞，會限制應用程序的吞吐量。

因此，在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量。

ConsumerRebalanceListener需要實現的兩個方法

下面的例子演示如何在失去partition的所有權之前通過onPartitionRevoked()方法來提交偏移量。

Consumer有個Rebalance的特性，即重新負載均衡，該特性依賴於一個協調器來實現。每當Consumer Group中有Consumer退出或有新的Consumer加入都會觸發Rebalance。

之所以要重新負載均衡，是為了將退出的Consumer所負責處理的數據再重新分配到組內的其他Consumer上進行處理。或當有新加入的Consumer時，將組內其他Consumer的負載壓力，重新進均勻分配，而不會說新加入一個Consumer就閑在那。

下面就用幾張圖簡單描述一下，各種情況觸發Rebalance時，組內成員是如何與協調器進行交互的。

Tips ：圖中的Coordinator是協調器，而generation則類似於樂觀鎖中的版本號，每當成員入組成功就會更新，也是起到一個並發控制的作用。

參考：

消息中間件Kafka – PHP操作使用Kafka

cd librdkafka/

./configure make make install

安裝成功界面 沒有報錯就是安裝成功

大數據Kafka有哪些優勢能力呢？

Kafka的高吞吐能力、緩存機制能有效的解決高峰流量衝擊問題。實踐表明，在未將kafka引入系統前，當互聯網關發送的數據量較大時，往往會掛起關係數據庫，數據常常丟失。在引入kafka後，更新程序能夠結合能力自主處理消息，不會引起數據丟失，關係型數據庫的壓力波動不會發生過於顯著的變化，不會出現數據庫掛起鎖死現象。

依靠kafka的訂閱分發機制，實現了一次發佈，各分支依據需求自主訂閱的功能。避免了各分支機構直接向數據中心請求數據，或者數據中心依次批量向分支機構傳輸數據以致實時性不足的情況。kafka提高了實時性，減輕了數據中心的壓力，提高了效率。為了幫助大家讓學習變得輕鬆、高效，給大家免費分享一大批資料，幫助大家在成為大數據工程師，乃至架構師的路上披荊斬棘。在這裡給大家推薦一個大數據學習交流圈：658558542 歡迎大家進群交流討論，學習交流，共同進步。

當真正開始學習的時候難免不知道從哪入手，導致效率低下影響繼續學習的信心。

但最重要的是不知道哪些技術需要重點掌握，學習時頻繁踩坑，最終浪費大量時間，所以有有效資源還是很有必要的。

消費者是以consumer group消費者組的方式工作，由一個或者多個消費者組成一個組，共同消費一個topic。每個分區在同一時間只能由group中的一個消費者讀取，但是多個group可以同時消費這個partition。在圖中，有一個由三個消費者組成的group，有一個消費者讀取主題中的兩個分區，另外兩個分別讀取一個分區。某個消費者讀取某個分區，也可以叫做某個消費者是某個分區的擁有者。

在這種情況下，消費者可以通過水平擴展的方式同時讀取大量的消息。另外，如果一個消費者失敗了，那麼其他的group成員會自動負載均衡讀取之前失敗的消費者讀取的分區。

消費方式

consumer採用pull（拉）模式從broker中讀取數據。

push（推）模式很難適應消費速率不同的消費者，因為消息發送速率是由broker決定的。它的目標是儘可能以最快速度傳遞消息，但是這樣很容易造成consumer來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。而pull模式則可以根據consumer的消費能力以適當的速率消費消息。

對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費消息的速率，同時consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的傳輸語義。

pull模式不足之處是，如果kafka沒有數據，消費者可能會陷入循環中，一直等待數據到達。為了避免這種情況，我們在我們的拉請求中有參數，允許消費者請求在等待數據到達的「長輪詢」中進行阻塞（並且可選地等待到給定的位元組數，以確保大的傳輸大小）。

消費者組的偏移量等信息存儲在zookeeper中的consumers節點中。

6.1 Kafka Producer 壓力測試

record-size 是一條信息有多大，單位是位元組。

num-records 是總共發送多少條信息。

throughput 是每秒多少條信息，設成-1，表示不限流，可測出生產者最大吞吐量。

大數據Kafka是什麼呢？

1.1　概述

Kafka是最初由Linkedin公司開發，是一個分佈式、分區的、多副本的、多訂閱者，基於zookeeper協調的分佈式日誌系統（也可以當做MQ系統），常見可以用於web/nginx日誌、訪問日誌，消息服務等等，Linkedin於2010年貢獻給了Apache基金會並成為頂級開源項目。

主要應用場景是：日誌收集系統和消息系統。

Kafka主要設計目標如下：

以時間複雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數據也能保證常數時間的訪問性能。

高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條消息的傳輸。

支持Kafka Server間的消息分區，及分佈式消費，同時保證每個partition內的消息順序傳輸。

同時支持離線數據處理和實時數據處理。

Scale out:支持在線水平擴展

二、Kafka的優點

2.1　解耦

在項目啟動之初來預測將來項目會碰到什麼需求，是極其困難的。消息系統在處理過程中間插入了一個隱含的、基於數據的接口層，兩邊的處理過程都要實現這一接口。這允許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。

2.2　冗餘（副本）

有些情況下，處理數據的過程會失敗。除非數據被持久化，否則將造成丟失。消息隊列把數據進行持久化直到它們已經被完全處理，通過這一方式規避了數據丟失風險。許多消息隊列所採用的”插入-獲取-刪除”範式中，在把一個消息從隊列中刪除之前，需要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。

2.3　擴展性

因為消息隊列解耦了你的處理過程，所以增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。不需要改變代碼、不需要調節參數。擴展就像調大電力按鈕一樣簡單。

2.4　靈活性峰值處理能力

在訪問量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量並不常見；如果為以能處理這類峰值訪問為標準來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵組件頂住突發的訪問壓力，而不會因為突發的超負荷的請求而完全崩潰。

2.5　可恢復性

系統的一部分組件失效時，不會影響到整個系統。消息隊列降低了進程間的耦合度，所以即使一個處理消息的進程掛掉，加入隊列中的消息仍然可以在系統恢復後被處理。

2.6　順序保證

在大多使用場景下，數據處理的順序都很重要。大部分消息隊列本來就是排序的，並且能保證數據會按照特定的順序來處理。Kafka保證一個Partition內的消息的有序性。

2.7　緩衝

在任何重要的系統中，都會有需要不同的處理時間的元素。例如，加載一張圖片比應用過濾器花費更少的時間。消息隊列通過一個緩衝層來幫助任務最高效率的執行———寫入隊列的處理會儘可能的快速。該緩衝有助於控制和優化數據流經過系統的速度。

2.8　異步通信

很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但並不立即處理它。想向隊列中放入多少消息就放多少，然後在需要的時候再去處理它們。

.2　消息系統介紹

一個消息系統負責將數據從一個應用傳遞到另外一個應用，應用只需關注於數據，無需關注數據在兩個或多個應用間是如何傳遞的。分佈式消息傳遞基於可靠的消息隊列，在客戶端應用和消息系統之間異步傳遞消息。有兩種主要的消息傳遞模式：點對點傳遞模式、發佈-訂閱模式。大部分的消息系統選用發佈-訂閱模式。Kafka就是一種發佈-訂閱模式。

1.3　點對點消息傳遞模式

1.4　發佈-訂閱消息傳遞模式

在發佈-訂閱消息系統中，消息被持久化到一個topic中。與點對點消息系統不同的是，消費者可以訂閱一個或多個topic，消費者可以消費該topic中所有的數據，同一條數據可以被多個消費者消費，數據被消費後不會立馬刪除。在發佈-訂閱消息系統中，消息的生產者稱為發佈者，消費者稱為訂閱者。該模式的示例圖如下：

大型的PHP應用，通常使用什麼應用做消息隊列？

一、消息隊列概述\x0d\x0a消息隊列中間件是分佈式系統中重要的組件，主要解決應用耦合，異步消息，流量削鋒等問題。實現高性能，高可用，可伸縮和最終一致性架構。是大型分佈式系統不可缺少的中間件。\x0d\x0a目前在生產環境，使用較多的消息隊列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。\x0d\x0a二、消息隊列應用場景\x0d\x0a以下介紹消息隊列在實際應用中常用的使用場景。異步處理，應用解耦，流量削鋒和消息通訊四個場景。\x0d\x0a2.1異步處理\x0d\x0a場景說明：用戶註冊後，需要發註冊郵件和註冊短訊。傳統的做法有兩種1.串行的方式；2.並行方式。\x0d\x0a（1）串行方式：將註冊信息寫入數據庫成功後，發送註冊郵件，再發送註冊短訊。以上三個任務全部完成後，返回給客戶端。（架構KKQ：466097527，歡迎加入）\x0d\x0a（2）並行方式：將註冊信息寫入數據庫成功後，發送註冊郵件的同時，發送註冊短訊。以上三個任務完成後，返回給客戶端。與串行的差別是，並行的方式可以提高處理的時間。\x0d\x0a假設三個業務節點每個使用50毫秒鐘，不考慮網絡等其他開銷，則串行方式的時間是150毫秒，並行的時間可能是100毫秒。\x0d\x0a因為CPU在單位時間內處理的請求數是一定的，假設CPU1秒內吞吐量是100次。則串行方式1秒內CPU可處理的請求量是7次（1000/150）。並行方式處理的請求量是10次（1000/100）。\x0d\x0a小結：如以上案例描述，傳統的方式系統的性能（並發量，吞吐量，響應時間）會有瓶頸。如何解決這個問題呢？\x0d\x0a引入消息隊列，將不是必須的業務邏輯，異步處理。改造後的架構如下：\x0d\x0a按照以上約定，用戶的響應時間相當於是註冊信息寫入數據庫的時間，也就是50毫秒。註冊郵件，發送短訊寫入消息隊列後，直接返回，因此寫入消息隊列的速度很快，基本可以忽略，因此用戶的響應時間可能是50毫秒。因此架構改變後，系統的吞吐量提高到每秒20 QPS。比串行提高了3倍，比並行提高了兩倍。\x0d\x0a2.2應用解耦\x0d\x0a場景說明：用戶下單後，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口。如下圖：\x0d\x0a傳統模式的缺點：\x0d\x0a1） 假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗；\x0d\x0a2） 訂單系統與庫存系統耦合；\x0d\x0a如何解決以上問題呢？引入應用消息隊列後的方案，如下圖：\x0d\x0a訂單系統：用戶下單後，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功。\x0d\x0a庫存系統：訂閱下單的消息，採用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作。\x0d\x0a假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單後，訂單系統寫入消息隊列就不再關心其他的後續操作了。實現訂單系統與庫存系統的應用解耦。\x0d\x0a2.3流量削鋒\x0d\x0a流量削鋒也是消息隊列中的常用場景，一般在秒殺或團搶活動中使用廣泛。\x0d\x0a應用場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。為解決這個問題，一般需要在應用前端加入消息隊列。\x0d\x0a可以控制活動的人數；\x0d\x0a可以緩解短時間內高流量壓垮應用；\x0d\x0a用戶的請求，服務器接收後，首先寫入消息隊列。假如消息隊列長度超過最大數量，則直接拋棄用戶請求或跳轉到錯誤頁面；\x0d\x0a秒殺業務根據消息隊列中的請求信息，再做後續處理。\x0d\x0a2.4日誌處理\x0d\x0a日誌處理是指將消息隊列用在日誌處理中，比如Kafka的應用，解決大量日誌傳輸的問題。架構簡化如下：\x0d\x0a日誌採集客戶端，負責日誌數據採集，定時寫受寫入Kafka隊列；\x0d\x0aKafka消息隊列，負責日誌數據的接收，存儲和轉發；\x0d\x0a日誌處理應用：訂閱並消費kafka隊列中的日誌數據；\x0d\x0a以下是新浪kafka日誌處理應用案例：\x0d\x0a(1)Kafka：接收用戶日誌的消息隊列。\x0d\x0a(2)Logstash：做日誌解析，統一成JSON輸出給Elasticsearch。\x0d\x0a(3)Elasticsearch：實時日誌分析服務的核心技術，一個schemaless，實時的數據存儲服務，通過index組織數據，兼具強大的搜索和統計功能。\x0d\x0a(4)Kibana：基於Elasticsearch的數據可視化組件，超強的數據可視化能力是眾多公司選擇ELK stack的重要原因。\x0d\x0a2.5消息通訊\x0d\x0a消息通訊是指，消息隊列一般都內置了高效的通信機制，因此也可以用在純的消息通訊。比如實現點對點消息隊列，或者聊天室等。\x0d\x0a點對點通訊：\x0d\x0a客戶端A和客戶端B使用同一隊列，進行消息通訊。\x0d\x0a聊天室通訊：\x0d\x0a客戶端A，客戶端B，客戶端N訂閱同一主題，進行消息發佈和接收。實現類似聊天室效果。\x0d\x0a以上實際是消息隊列的兩種消息模式，點對點或發佈訂閱模式。模型為示意圖，供參考。\x0d\x0a三、消息中間件示例\x0d\x0a3.1電商系統\x0d\x0a消息隊列採用高可用，可持久化的消息中間件。比如Active MQ，Rabbit MQ，Rocket Mq。（1）應用將主幹邏輯處理完成後，寫入消息隊列。消息發送是否成功可以開啟消息的確認模式。（消息隊列返回消息接收成功狀態後，應用再返回，這樣保障消息的完整性）\x0d\x0a（2）擴展流程（發短訊，配送處理）訂閱隊列消息。採用推或拉的方式獲取消息並處理。\x0d\x0a（3）消息將應用解耦的同時，帶來了數據一致性問題，可以採用最終一致性方式解決。比如主數據寫入數據庫，擴展應用根據消息隊列，並結合數據庫方式實現基於消息隊列的後續處理。\x0d\x0a3.2日誌收集系統\x0d\x0a分為Zookeeper註冊中心，日誌收集客戶端，Kafka集群和Storm集群（OtherApp）四部分組成。\x0d\x0aZookeeper註冊中心，提出負載均衡和地址查找服務；\x0d\x0a日誌收集客戶端，用於採集應用系統的日誌，並將數據推送到kafka隊列；\x0d\x0a四、JMS消息服務\x0d\x0a講消息隊列就不得不提JMS 。JMS（Java Message Service,Java消息服務）API是一個消息服務的標準/規範，允許應用程序組件基於JavaEE平台創建、發送、接收和讀取消息。它使分佈式通信耦合度更低，消息服務更加可靠以及異步性。\x0d\x0a在EJB架構中，有消息bean可以無縫的與JM消息服務集成。在J2EE架構模式中，有消息服務者模式，用於實現消息與應用直接的解耦。\x0d\x0a4.1消息模型\x0d\x0a在JMS標準中，有兩種消息模型P2P（Point to Point）,Publish/Subscribe(Pub/Sub)。\x0d\x0a4.1.1 P2P模式\x0d\x0aP2P模式包含三個角色：消息隊列（Queue），發送者(Sender)，接收者(Receiver)。每個消息都被發送到一個特定的隊列，接收者從隊列中獲取消息。隊列保留着消息，直到他們被消費或超時。\x0d\x0aP2P的特點\x0d\x0a每個消息只有一個消費者（Consumer）(即一旦被消費，消息就不再在消息隊列中)\x0d\x0a發送者和接收者之間在時間上沒有依賴性，也就是說當發送者發送了消息之後，不管接收者有沒有正在運行，它不會影響到消息被發送到隊列\x0d\x0a接收者在成功接收消息之後需向隊列應答成功\x0d\x0a如果希望發送的每個消息都會被成功處理的話，那麼需要P2P模式。（架構KKQ：466097527，歡迎加入）\x0d\x0a4.1.2 Pub/sub模式\x0d\x0a包含三個角色主題（Topic），發佈者（Publisher），訂閱者（Subscriber） 。多個發佈者將消息發送到Topic,系統將這些消息傳遞給多個訂閱者。\x0d\x0aPub/Sub的特點\x0d\x0a每個消息可以有多個消費者\x0d\x0a發佈者和訂閱者之間有時間上的依賴性。針對某個主題（Topic）的訂閱者，它必須創建一個訂閱者之後，才能消費發佈者的消息。\x0d\x0a為了消費消息，訂閱者必須保持運行的狀態。\x0d\x0a為了緩和這樣嚴格的時間相關性，JMS允許訂閱者創建一個可持久化的訂閱。這樣，即使訂閱者沒有被激活（運行），它也能接收到發佈者的消息。\x0d\x0a如果希望發送的消息可以不被做任何處理、或者只被一個消息者處理、或者可以被多個消費者處理的話，那麼可以採用Pub/Sub模型。\x0d\x0a4.2消息消費\x0d\x0a在JMS中，消息的產生和消費都是異步的。對於消費來說，JMS的消息者可以通過兩種方式來消費消息。\x0d\x0a（1）同步\x0d\x0a訂閱者或接收者通過receive方法來接收消息，receive方法在接收到消息之前（或超時之前）將一直阻塞；\x0d\x0a（2）異步\x0d\x0a訂閱者或接收者可以註冊為一個消息監聽器。當消息到達之後，系統自動調用監聽器的onMessage方法。\x0d\x0aJNDI：Java命名和目錄接口,是一種標準的Java命名系統接口。可以在網絡上查找和訪問服務。通過指定一個資源名稱，該名稱對應於數據庫或命名服務中的一個記錄，同時返回資源連接建立所必須的信息。\x0d\x0aJNDI在JMS中起到查找和訪問發送目標或消息來源的作用。（架構KKQ：466097527，歡迎加入）\x0d\x0a4.3JMS編程模型\x0d\x0a(1) ConnectionFactory\x0d\x0a創建Connection對象的工廠，針對兩種不同的jms消息模型，分別有QueueConnectionFactory和TopicConnectionFactory兩種。可以通過JNDI來查找ConnectionFactory對象。\x0d\x0a(2) Destination\x0d\x0aDestination的意思是消息生產者的消息發送目標或者說消息消費者的消息來源。對於消息生產者來說，它的Destination是某個隊列（Queue）或某個主題（Topic）;對於消息消費者來說，它的Destination也是某個隊列或主題（即消息來源）。\x0d\x0a所以，Destination實際上就是兩種類型的對象：Queue、Topic可以通過JNDI來查找Destination。\x0d\x0a(3) Connection\x0d\x0aConnection表示在客戶端和JMS系統之間建立的鏈接（對TCP/IP socket的包裝）。Connection可以產生一個或多個Session。跟ConnectionFactory一樣，Connection也有兩種類型：QueueConnection和TopicConnection。\x0d\x0a(4) Session\x0d\x0aSession是操作消息的接口。可以通過session創建生產者、消費者、消息等。Session提供了事務的功能。當需要使用session發送/接收多個消息時，可以將這些發送/接收動作放到一個事務中。同樣，也分QueueSession和TopicSession。\x0d\x0a(5) 消息的生產者\x0d\x0a消息生產者由Session創建，並用於將消息發送到Destination。同樣，消息生產者分兩種類型：QueueSender和TopicPublisher。可以調用消息生產者的方法（send或publish方法）發送消息。\x0d\x0a(6) 消息消費者\x0d\x0a消息消費者由Session創建，用於接收被發送到Destination的消息。兩種類型：QueueReceiver和TopicSubscriber。可分別通過session的createReceiver(Queue)或createSubscriber(Topic)來創建。當然，也可以session的creatDurableSubscriber方法來創建持久化的訂閱者。\x0d\x0a(7) MessageListener\x0d\x0a消息監聽器。如果註冊了消息監聽器，一旦消息到達，將自動調用監聽器的onMessage方法。EJB中的MDB（Message-Driven Bean）就是一種MessageListener。\x0d\x0a深入學習JMS對掌握JAVA架構，EJB架構有很好的幫助，消息中間件也是大型分佈式系統必須的組件。本次分享主要做全局性介紹，具體的深入需要大家學習，實踐，總結，領會。\x0d\x0a五、常用消息隊列\x0d\x0a一般商用的容器，比如WebLogic，JBoss，都支持JMS標準，開發上很方便。但免費的比如Tomcat，Jetty等則需要使用第三方的消息中間件。本部分內容介紹常用的消息中間件（Active MQ,Rabbit MQ，Zero MQ,Kafka）以及他們的特點。\x0d\x0a5.1 ActiveMQ\x0d\x0aActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規範的 JMS Provider實現，儘管JMS規範出台已經是很久的事情了，但是JMS在當今的J2EE應用中間仍然扮演着特殊的地位。\x0d\x0aActiveMQ特性如下：\x0d\x0a⒈ 多種語言和協議編寫客戶端。語言: Java,C,C++,C#,Ruby,Perl,Python,PHP。應用協議： OpenWire,Stomp REST,WS Notification,XMPP,AMQP\x0d\x0a⒉ 完全支持JMS1.1和J2EE 1.4規範 （持久化，XA消息，事務)\x0d\x0a⒊ 對spring的支持，ActiveMQ可以很容易內嵌到使用Spring的系統裏面去，而且也支持Spring2.0的特性\x0d\x0a⒋ 通過了常見J2EE服務器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的測試，其中通過JCA 1.5 resource adaptors的配置，可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上\x0d\x0a⒌ 支持多種傳送協議：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA\x0d\x0a⒍ 支持通過JDBC和journal提供高速的消息持久化\x0d\x0a⒎ 從設計上保證了高性能的集群，客戶端-服務器，點對點\x0d\x0a⒏ 支持Ajax\x0d\x0a⒐ 支持與Axis的整合\x0d\x0a⒑ 可以很容易得調用內嵌JMS provider，進行測試\x0d\x0a5.2 RabbitMQ\x0d\x0aRabbitMQ是流行的開源消息隊列系統，用erlang語言開發。RabbitMQ是AMQP（高級消息隊列協議）的標準實現。支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX，持久化。用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。\x0d\x0a幾個重要概念：\x0d\x0aBroker：簡單來說就是消息隊列服務器實體。\x0d\x0aExchange：消息交換機，它指定消息按什麼規則，路由到哪個隊列。\x0d\x0aQueue：消息隊列載體，每個消息都會被投入到一個或多個隊列。\x0d\x0aBinding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。\x0d\x0aRouting Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。\x0d\x0avhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。\x0d\x0aproducer：消息生產者，就是投遞消息的程序。\x0d\x0aconsumer：消息消費者，就是接受消息的程序。\x0d\x0achannel：消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務。\x0d\x0a消息隊列的使用過程，如下：\x0d\x0a（1）客戶端連接到消息隊列服務器，打開一個channel。\x0d\x0a（2）客戶端聲明一個exchange，並設置相關屬性。\x0d\x0a（3）客戶端聲明一個queue，並設置相關屬性。\x0d\x0a（4）客戶端使用routing key，在exchange和queue之間建立好綁定關係。\x0d\x0a（5）客戶端投遞消息到exchange。\x0d\x0aexchange接收到消息後，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列里。\x0d\x0a5.3 ZeroMQ\x0d\x0a號稱史上最快的消息隊列，它實際類似於Socket的一系列接口，他跟Socket的區別是：普通的socket是端到端的（1:1的關係），而ZMQ卻是可以N：M 的關係，人們對BSD套接字的了解較多的是點對點的連接，點對點連接需要顯式地建立連接、銷毀連接、選擇協議（TCP/UDP）和處理錯誤等，而ZMQ屏蔽了這些細節，讓你的網絡編程更為簡單。ZMQ用於node與node間的通信，node可以是主機或者是進程。\x0d\x0a引用官方的說法： 「ZMQ(以下ZeroMQ簡稱ZMQ)是一個簡單好用的傳輸層，像框架一樣的一個socket library，他使得Socket編程更加簡單、簡潔和性能更高。是一個消息處理隊列庫，可在多個線程、內核和主機盒之間彈性伸縮。ZMQ的明確目標是「成為標準網絡協議棧的一部分，之後進入Linux內核」。現在還未看到它們的成功。但是，它無疑是極具前景的、並且是人們更加需要的「傳統」BSD套接字之上的一 層封裝。ZMQ讓編寫高性能網絡應用程序極為簡單和有趣。」\x0d\x0a特點是：\x0d\x0a高性能，非持久化；\x0d\x0a跨平台：支持Linux、Windows、OS X等。\x0d\x0a多語言支持； C、C++、Java、.NET、Python等30多種開發語言。\x0d\x0a可單獨部署或集成到應用中使用；\x0d\x0a可作為Socket通信庫使用。\x0d\x0a與RabbitMQ相比，ZMQ並不像是一個傳統意義上的消息隊列服務器，事實上，它也根本不是一個服務器，更像一個底層的網絡通訊庫，在Socket API之上做了一層封裝，將網絡通訊、進程通訊和線程通訊抽象為統一的API接口。支持「Request-Reply 「，」Publisher-Subscriber「，」Parallel Pipeline」三種基本模型和擴展模型。\x0d\x0aZeroMQ高性能設計要點：\x0d\x0a1、無鎖的隊列模型\x0d\x0a對於跨線程間的交互（用戶端和session）之間的數據交換通道pipe，採用無鎖的隊列算法CAS；在pipe兩端註冊有異步事件，在讀或者寫消息到pipe的時，會自動觸發讀寫事件。\x0d\x0a2、批量處理的算法\x0d\x0a對於傳統的消息處理，每個消息在發送和接收的時候，都需要系統的調用，這樣對於大量的消息，系統的開銷比較大，zeroMQ對於批量的消息，進行了適應性的優化，可以批量的接收和發送消息。\x0d\x0a3、多核下的線程綁定，無須CPU切換\x0d\x0a區別於傳統的多線程併發模式，信號量或者臨界區， zeroMQ充分利用多核的優勢，每個核綁定運行一個工作者線程，避免多線程之間的CPU切換開銷。\x0d\x0a5.4 Kafka\x0d\x0aKafka是一種高吞吐量的分佈式發佈訂閱消息系統，它可以處理消費者規模的網站中的所有動作流數據。 這種動作（網頁瀏覽，搜索和其他用戶的行動）是在現代網絡上的許多社會功能的一個關鍵因素。 這些數據通常是由於吞吐量的要求而通過處理日誌和日誌聚合來解決。 對於像Hadoop的一樣的日誌數據和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。Kafka的目的是通過Hadoop的並行加載機制來統一線上和離線的消息處理，也是為了通過集群機來提供實時的消費。\x0d\x0aKafka是一種高吞吐量的分佈式發佈訂閱消息系統，有如下特性：\x0d\x0a通過O(1)的磁盤數據結構提供消息的持久化，這種結構對於即使數以TB的消息存儲也能夠保持長時間的穩定性能。（文件追加的方式寫入數據，過期的數據定期刪除）\x0d\x0a高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒數百萬的消息。\x0d\x0a支持通過Kafka服務器和消費機集群來分區消息。\x0d\x0a支持Hadoop並行數據加載。\x0d\x0aKafka相關概念\x0d\x0aBroker\x0d\x0aKafka集群包含一個或多個服務器，這種服務器被稱為broker[5]\x0d\x0aTopic\x0d\x0a每條發佈到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存於一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數據而不必關心數據存於何處）\x0d\x0aPartition\x0d\x0aParition是物理上的概念，每個Topic包含一個或多個Partition.\x0d\x0aProducer\x0d\x0a負責發佈消息到Kafka broker\x0d\x0aConsumer\x0d\x0a消息消費者，向Kafka broker讀取消息的客戶端。\x0d\x0aConsumer Group\x0d\x0a每個Consumer屬於一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬於默認的group）。\x0d\x0a一般應用在大數據日誌處理或對實時性（少量延遲），可靠性（少量丟數據）要求稍低的場景使用。

Kafka使用場景

Kafka作為一個傳統的消息代理的替代品表現得非常出色。使用消息代理有各種各樣的原因(將處理與數據生成器解耦，緩衝未處理的消息，等等)。與大多數消息傳遞系統相比，Kafka有更好的吞吐量、內置分區、複製和容錯性，這使得它成為大規模消息處理應用的一個很好的解決方案。

根據我們的經驗，消息傳遞的使用通常是相對較低的吞吐量，但可能需要較低的端到端延遲，並且常常依賴於Kafka提供的強大的持久性保證。

在這個領域，Kafka可以與ActiveMQ或RabbitMQ等傳統消息傳遞系統相媲美。

Kafka最初的用例是能夠重建一個用戶活動跟蹤管道，作為一組實時發佈-訂閱提要。這意味着站點活動(頁面瀏覽、搜索或用戶可能採取的其他操作)被發佈到中心主題，每個活動類型有一個主題。這些提要可用於訂閱一系列用例，包括實時處理、實時監視和加載到Hadoop或脫機數據倉庫系統以進行脫機處理和報告。

活動跟蹤通常是非常大的量，因為許多活動消息會生成的每個用戶頁面視圖。

Kafka通常用於運行監控數據。這涉及聚合來自分佈式應用程序的統計信息，以生成集中的操作數據提要。

許多人使用Kafka作為日誌聚合解決方案的替代品。日誌聚合通常收集服務器上的物理日誌文件，並將它們放在一個中心位置(可能是文件服務器或HDFS)進行處理。Kafka抽象了文件的細節，並以消息流的形式對日誌或事件數據進行了更清晰的抽象。這允許低延遲處理，並更容易支持多個數據源和分佈式數據消費。與以日誌為中心的系統如Scribe或Flume相比，Kafka提供了同樣好的性能，由於複製而更強的持久性保證，以及更低的端到端延遲。

很多Kafka的用戶在處理數據的管道中都有多個階段，原始的輸入數據會從Kafka的主題中被消費，然後被聚合、充實或者轉換成新的主題進行進一步的消費或者後續的處理。例如，推薦新聞文章的處理管道可能會從RSS源抓取文章內容，並將其發佈到「文章」主題;進一步的處理可能會規範化或刪除該內容，並將清理後的文章內容發佈到新主題;最後一個處理階段可能會嘗試向用戶推薦這些內容。這種處理管道基於單個主題創建實時數據流圖。從0.10.0.0開始，Apache Kafka提供了一個輕量級但功能強大的流處理庫，名為Kafka Streams，用於執行上述的數據處理。除了Kafka Streams，其他開源流處理工具包括Apache Storm和Apache Samza。

事件溯源是一種應用程序設計風格，其中將狀態更改記錄為按時間順序排列的記錄序列。Kafka支持非常大的存儲日誌數據，這使得它成為這種風格的應用程序的優秀後端。

Kafka可以作為分佈式系統的一種外部提交日誌。日誌有助於在節點之間複製數據，並充當故障節點的重新同步機制，以恢復它們的數據。Kafka的日誌壓縮特性支持這種用法。在這種用法中，Kafka類似於Apache BookKeeper項目。