02 消息訂閱與發布kafka(kafka入門：一個開源的、輕量級、高吞吐、高可用的分布式消息系統)

如果客戶端沒有指定消費者的id， Kafka會自動為該消費者生成一個全局唯一的id，格式為${ groupId}-${ hostName}-${ timestamp}-${ UUID前8位字符}。 同一個主題的一條消息只能被同一個消費組下某一個消費者消費 ，但不同消費組的消費者可同時消費該消息。 消費組是Kafka用來實現對一個主題消息進行廣播和單播的手段 ，實現消息廣播只需指定各消費者均屬于不同的消費組，消息單播則只需讓各消費者屬于同一個消費組。

推論： kafka消息是按照消息類型(主題)，在一個消費者組中只能消費一次。也就是一個消費者組只消費一類型的消息。如果某個服務要消費一類消息，必須將自己置為不同的消費者組。

Kafka利用ZooKeeper保存相應元數據信息， Kafka元數據信息包括如代理節點信息、Kafka集群信息、舊版消費者信息及其消費偏移量信息、主題信息、分區狀態信息、分區副本分配方案信息、動態配置信息等。 Kafka在啟動或運行過程當中會在ZooKeeper上創建相應節點 來保存元數據信息， Kafka通過監聽機制在這些節點注冊相應監聽器來監聽節點元數據的變化 ，從而由ZooKeeper負責管理維護Kafka集群，同時通過ZooKeeper我們能夠很方便地對Kafka集群進行水平擴展及數據遷移。

消息隊列之zeroMQ、rabbitMQ、kafka

      首先消息是網絡通訊的載體，隊列可以理解是一種先進先出的數據結構，消息隊列是存放消息的容器，是分布式系統中的重要組件。消息隊列的優勢在于：解耦、異步、削峰，把相關性不

強的模塊獨立分開視為解耦，異步就是非必要邏輯異步方式處理，加快響應速度，削峰是避免短期高并發導致系統問題進行緩沖隊列處理。消息隊列的缺點在于：加強系統復雜性、系統可用性降低，使

用了消息隊列系統出現問題排查的范圍就變大、需要考慮消息隊列導致的問題。

          本文說明主流的消息隊列，針對使用過的zeroMQ和rabbitMQ、Kakfa：

          zeroMQ ：C語言開發，號稱最快的消息隊列，本著命名zero的含義，中油中間架構使用簡單，表面上是基于socket的封裝套接字API，在多個節點應用場景下非常靈活、架構的可擴展性很強，

實現N到M的協同處理；

            zmq的socket模式： req、rep、push、pull、pub、sub、router、dealer。

          （1）req和rep：請求回應模型，req和rep都可以請求和回答，不同的只是req是先send再rec，rep是先rec再send。支持N個請求端一個接收端，也支持N個接收端一個請求端。N個接收端采

用rr負載均衡。 哪個是“一”端，哪個就bind端口，“N”端就只能買粉絲nnect，所以，req+rep無論誰bind端口，肯定要有一個是“一”。

          （2）  router和dealer：隨時可以發送和接收的req和rep，看起來router+dealer跟 req+rep屬于同類功能。因為router和dealer可以隨時發送接收，所以它們可以用來做路由。一個router用來響

應N個req，然后它在響應處理的時候，再通過另一個socket把請求扔出去，接收者是另外的M個rep，這就做到N:M。

         （3）pub和sub ：訂閱和推送，對應發布者和訂閱者。

         （4）push和pull：就是管道，一個只推數據，一個只拉數據。

           rabbitMQ ：使用erlang語言開發，高并發特點，基于AMQP（即Advanced Message Queuing Proto買粉絲l）的開源高級消費隊列，AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/

訂閱）、可靠性、安全），企業級適應性和穩定性，并且有WEB管理界面方便用戶查看和管理。以下是rabbitMQ的結構圖：

         （1）Procer:數據發送方，一般一個Message有兩個部分：payload（有效載荷）和label（標簽），payload是數據實際載體，label是exchange的名字或者一個tag，決定發給哪個Consumer；

         （2）Exchange: 內部 消息交換器,exchange從生產者那收到消息后，一般會指定一個Routing Key，來指定這個消息的路由規則，當然Routing Key需要與Exchange Type及Binding key聯合使用

才能最終生效，根據路由規則，匹配查詢表中的routing key，分發消息到queue中；

         （3）binding：即綁定，綁定（Binding）Exchange與Queue的同時，一般會指定一個Binding key，但不一定會生效，依賴于Exchange Type；

         （4）Queue：即隊列是rabbitmq內部對象，用于存儲消息，一個message可以被同時拷貝到多個queue中，queue對load balance的處理是完美的。對于多個Consumer來說，RabbitMQ 使用循

環的方式（round-robin）的方式均衡的發送給不同的Consumer；

         （5）Connection與Channel： Connection 就是一個TCP的連接，Procer和Consumer都是通過TCP連接到RabbitMQ Server， Channel 是為了節省開銷建立在上述的TCP連接中的接口，大部

分的業務操作是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發布消息等；

        （6）Consumer：即數據的接收方，如果有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者；

        （7）Broker: 即RabbitMQ Server，其作用是維護一條從Procer到Consumer的路線，保證數據能夠按照指定的方式進行傳輸；

       （8）Virtual host：即虛擬主機，當多個不同的用戶使用同一個RabbitMQ server提供的服務時，可以劃分出多個vhost，每個用戶在自己的vhost創建exchange／queue；

         rabbitMQ消息轉發中的路由轉發是重點，生產者Procer在發送消息時，都需要指定一個RoutingKey和Exchange，Exchange收到消息后可以看到消息中指定的RoutingKey，再根據當前

Exchange的ExchangeType,按一定的規則將消息轉發到相應的queue中去。三種Exchage type：

       （1）Direct exchange ：直接轉發路由，原理是通過消息中的routing key，與binding 中的binding-key 進行比對，若二者匹配，則將消息發送到這個消息隊列；

          比如：消息生成者生成一個message(payload是1，routing key為蘋果)，兩個binding(binding key分別為蘋果、香蕉);exchange比對消息的routing key和binding key后，將消息發給了queue1，消息消費者1獲得queue1的消息；

       （2）Topic exchange： 通配路由，是direct exchange的通配符模式，

          比如：消息生成者生成一個message(payload是1，routing key為quick.orange.rabbit)，兩個binding(binding key分別為*.orange. 、 *.*.rabbit)；exchange比對消息的routing key和binding key

后,exchange將消息分發給兩個queue，兩個消費者獲得queue的消息；

     （3）Fanout exchange： 復制分發路由，原理是不需要routkey，當exchange收到消息后，將消息復制多份轉發給與自己綁定的消息隊列，

          比如：消息生成者生成一個message(payload是1，routing key為蘋果)，兩個binding(binding key分別為蘋果、香蕉)；exchange將消息分發給兩個queue，兩個消費者獲得queue的消息；

       rabbiMQ如何保證消息的可靠性？

     （1）Message rability：消息持久化，非持久化消息保存在內存中，持久化消息寫入內存同時也寫入磁盤；

     （2）Message acknowledgment：消息確認機制，可以要求消費者在消費完消息后發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）后才將該消息從Queue中移

除。通過ACK。每個Message都要被acknowledged（確認，ACK）。

     （3）生產者消息確認機制：AMQP事務機制、生產者消息確認機制（publisher 買粉絲nfirm）。

     最后， 對比一下zeroMQ、rabbitMQ、kafka主流的消息隊列

很赞哦!（5）