一、前言

     上一篇部落格介紹了用執行緒池實現非同步任務。這一篇部落格談一談用MQ實現非同步任務。MQ的產品有灰常多，像什麼MSMQ、activeMQ、RocketMQ、RabbitMQ、kafak等。在此之前先談一談對訊息佇列的理解。

二、MQ

       MQ是一種應用程式對應用程式的通訊方法，應用程式通過讀寫出入佇列的訊息來進行通訊，兩者無需建立連線，釋出者和消費者無需知道對方的存在。

      MQ是生產者--消費者模型的一個代表，一端往訊息佇列中不斷寫入訊息，而另一端則可以不斷讀取或訂閱佇列中的訊息。

      使用場景：

      1、非同步處理：在專案中將一些無需及時返回且耗時的操作提取出來，進行非同步處理，採用非同步處理，將大大節省了伺服器的請求響應時間，從而提高了系統的吞吐量。

       2、訊息推送

三、RabbitMQ特點

       RabbitMQ是一個由Erlang語言開發的AMQP的開源實現。

       RabbitMQ的特點為：

       1、可靠性

       RabbitMQ提供了多種技術可以讓你在效能和可靠性之間進行權衡，如永續性、傳輸確認、投遞確認。

       2、靈活的路由

       在訊息進行佇列之前，通過Exchange來路由訊息。對於典型的路由功能，RabbitMQ已經提供了一些內建的Exchange來實現，針對更復雜的路由功能，可以將多個Exchange繫結在一起，也可以通過外掛機制來實現自己的Exchange。

       3、訊息叢集

        多個RabbitMQ伺服器可以組成一個叢集，形成一個邏輯Broker。

       4、高可用

       在同一叢集中佇列可以被映象到多臺機器，使得在部分節點出現問題的情況下，佇列仍然可用。

       5、多協議

       RabbitMQ支援多種訊息佇列協議。

       6、多語言客戶端

       支援java、.NET、Ruby等

       7、管理介面

        RabbitMQ提供了一個易用的使用者介面，使用者可以監控和管理Broker的許多方面。

       8、跟蹤機制

        如果訊息異常，RabbitMQ提供了訊息跟蹤機制，使用者可以找出發生了什麼。

       9、外掛機制

        RabbitMQ提供了很多外掛，來從多方面進行擴充套件，也可以編寫各種外掛。

四、RabbitMQ中的概念模型

•  訊息抽象模型

       消費者訂閱某個佇列，生產者建立訊息，然後釋出到佇列中，最後將訊息傳送到監聽的消費者。

• RabbitMQ的基本概念

   上圖是一個最簡單的抽象模型，接下來將展示一個更加詳細的模型。RabbitMQ內部結構如下：

   1、Message

    Message由訊息頭和訊息體組成，訊息體是不透明的，訊息頭是由一些屬性組成，這些屬性包括routing-key（路由鍵）、priority（相對於其他訊息的優先權）、delivery-mode（指出該訊息可能需要永續性儲存）等。

   2、Publisher

    訊息的生產者，也是一個向交換機發布訊息的客戶端應用程式。

   3、Exchange

    用來接收生產者傳送的訊息，並按照一定的規則將這些訊息路由給伺服器中的佇列。

   4、Binding

    繫結，用於訊息佇列和交換器之間的路由關聯，一個繫結就是將交換器和佇列連線起來的路由規則。

   5、Queue

   用來儲存訊息直到傳送給消費者，它是訊息的容器，1個訊息可投入一個到多個佇列，訊息一直在佇列裡，等待消費者取走。

   6、Connection

    網路連線，比如一個TCP連線。

   7、Channel

    通道，或者稱之為管道，是一條雙向資料流通道，AMQP都是通過通道發出去的，不管是釋出訊息，訂閱佇列，還是接受訊息，這些動作都是通過通道來完成的。

    8、Consumer

    訊息的消費者，表示一個從訊息佇列中取的訊息的客戶端應用程式。

    9、Virtual Host

    表示一批交換器、訊息佇列和相關物件。每個 vhost 本質上就是一個 mini 版的 RabbitMQ 伺服器，擁有自己的佇列、交換器、繫結和許可權機制。vhost 是 AMQP 概念的基礎，必須在連線時指定，RabbitMQ 預設的 vhost 是 / 。

    10、Broker

     表示RabbitMQ伺服器實體

五、入門Demo

      pom檔案新增依賴

    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>4.3.0</version>
    </dependency>

    <dependency>
      <groupId>com.github.sstone</groupId>
      <artifactId>amqp-client_2.10</artifactId>
      <version>1.3</version>
    </dependency>

1、Publisher

public class Publisher {
    //佇列名稱
    private final static String QUEUE_NAME = "Queue";
    public static void main(String[] args) {
        // 建立連線工廠
        ConnectionFactory factory = null;
        // 建立到代理伺服器到連線
        Connection connection = null;
        // 獲得通道
        Channel channel = null;
        try {
            factory = new ConnectionFactory();
            //設定使用者名稱和密碼
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 設定 RabbitMQ 地址
            factory.setHost("localhost");
            // 建立到代理伺服器到連線
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "hello world，hello world";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
            System.out.println("傳送  message[" + message + "] to "+ QUEUE_NAME +" success!");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            try {
// 關閉資源
                channel.close();
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }
}

2、consumer

public class Consumer {
    //佇列名稱
    private final static String QUEUE_NAME = "Queue";
    public static void main(String[] args) {
        // 建立連線工廠
        ConnectionFactory factory = null;
        // 建立到代理伺服器到連線
        Connection connection = null;
        // 獲得通道
        Channel channel = null;
        try {
            factory = new ConnectionFactory();
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setHost("localhost");
            // 建立到代理伺服器到連線
            connection = factory.newConnection();
            channel = connection.createChannel();
            // 1.佇列名2.是否持久化，3是否侷限與連結，4不再使用是否刪除，5其他的屬性
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 宣告一個消費者,配置好獲取訊息的方式
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(QUEUE_NAME, true, consumer);
            // 迴圈獲取訊息
            while (true) {
            // 迴圈獲取資訊
            // 指向下一個訊息，如果沒有會一直阻塞
                Delivery delivery = consumer.nextDelivery();
                String msg = new String(delivery.getBody());
                System.out.println("接收 message[" + msg + "] from " + QUEUE_NAME);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (ShutdownSignalException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            try {
            // 關閉資源
                channel.close();
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }
}

效果圖如下：