消息隊列（MQ）

什么是消息隊列

消息隊列，即MQ，Message Queue。

消息隊列是典型的：生產者、消費者模型。生產者不斷向消息隊列中生產消息，消費者不斷的從隊列中獲取消息。因為消息的生產和消費都是異步的，而且只關心消息的發送和接收，沒有業務邏輯的侵入，這樣就實現了生產者和消費者的解耦。

結合前面所說的問題：

• 商品服務對商品增刪改以后，無需去操作索引庫或靜態頁面，只是發送一條消息，也不關心消息被誰接收。

• 搜索服務和靜態頁面服務接收消息，分別去處理索引庫和靜態頁面。

如果以后有其它系統也依賴商品服務的數據，同樣監聽消息即可，商品服務無需任何代碼修改。

 

AMQP和JMS

MQ是消息通信的模型，並發具體實現。現在實現MQ的有兩種主流方式：AMQP、JMS。

 

兩者間的區別和聯系：

• JMS是定義了統一的接口，來對消息操作進行統一；AMQP是通過規定協議來統一數據交互的格式

• JMS限定了必須使用Java語言；AMQP只是協議，不規定實現方式，因此是跨語言的。

• JMS規定了兩種消息模型；而AMQP的消息模型更加豐富

常見MQ產品

 

 

• ActiveMQ：基於JMS， Apache

• RabbitMQ：基於AMQP協議，erlang語言開發，穩定性好

• RocketMQ：基於JMS，阿里巴巴產品，目前交由Apache基金會

• Kafka：分布式消息系統，高吞吐量

RabbitMQ

RabbitMQ是基於AMQP的一款消息管理系統

官網： http://www.rabbitmq.com/

官方教程：http://www.rabbitmq.com/getstarted.html

 

RabbitMQ基於Erlang語言開發：

下載和安裝

下載

官網下載地址：http://www.rabbitmq.com/download.html

 

 

我們使用的是：3.4.1版本

安裝

 

---

五種消息模型

RabbitMQ提供了6種消息模型，但是第6種其實是RPC，並不是MQ，因此不予學習。那么也就剩下5種。

但是其實3、4、5這三種都屬於訂閱模型，只不過進行路由的方式不同。

 

 

導入demo工程

我們通過一個demo工程來了解下RabbitMQ的工作方式：

 

 

基本消息模型

說明

官方文檔說明：

RabbitMQ是一個消息的代理者（Message Broker）：它接收消息並且傳遞消息。

你可以認為它是一個郵局：當你投遞郵件到一個郵箱，你很肯定郵遞員會終究會將郵件遞交給你的收件人。與此類似，RabbitMQ 可以是一個郵箱、郵局、同時還有郵遞員。

不同之處在於：RabbitMQ不是傳遞紙質郵件，而是二進制的數據

基本消息模型圖：

 

在上圖的模型中，有以下概念：

• P：生產者，也就是要發送消息的程序

• C：消費者：消息的接受者，會一直等待消息到來。

• queue：消息隊列，圖中紅色部分。類似一個郵箱，可以緩存消息；生產者向其中投遞消息，消費者從其中取出消息。

生產者

連接工具類：

public class ConnectionUtil {
    /**
     * 建立與RabbitMQ的連接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定義連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設置服務地址
        factory.setHost("192.168.56.101");
        //端口
        factory.setPort(5672);
        //設置賬號信息，用戶名、密碼、vhost
        factory.setVirtualHost("/leyou");
        factory.setUsername("leyou");
        factory.setPassword("leyou");
        // 通過工程獲取連接
        Connection connection = factory.newConnection();
        return connection;
    }
}

　　

生產者發送消息：

public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 從連接中創建通道，使用通道才能完成消息相關的操作
        Channel channel = connection.createChannel();
        // 聲明（創建）隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 消息內容
        String message = "Hello World!";
        // 向指定的隊列中發送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        
        System.out.println(" [x] Sent '" + message + "'");

        //關閉通道和連接
        channel.close();
        connection.close();
    }
}

　　

控制台：

 

web控制台查看消息

進入隊列頁面，可以看到新建了一個隊列：simple_queue

 

 

點擊隊列名稱，進入詳情頁，可以查看消息：

 

在控制台查看消息並不會將消息消費，所以消息還在。

消費者獲取消息

public class Recv {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 創建通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 監聽隊列，第二個參數：是否自動進行消息確認。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

控制台：

 

這個時候，隊列中的消息就沒了：

 

消費者的消息確認機制

通過剛才的案例可以看出，消息一旦被消費者接收，隊列中的消息就會被刪除。

那么問題來了：RabbitMQ怎么知道消息被接收了呢？

這就要通過消息確認機制（Acknowlege）來實現了。當消費者獲取消息后，會向RabbitMQ發送回執ACK，告知消息已經被接收。不過這種回執ACK分兩種情況：

• 自動ACK：消息一旦被接收，消費者自動發送ACK

• 手動ACK：消息接收后，不會發送ACK，需要手動調用

大家覺得哪種更好呢？

這需要看消息的重要性：

• 如果消息不太重要，丟失也沒有影響，那么自動ACK會比較方便

• 如果消息非常重要，不容丟失。那么最好在消費完成后手動ACK，否則接收消息后就自動ACK，RabbitMQ就會把消息從隊列中刪除。如果此時消費者宕機，那么消息就丟失了。

我們之前的測試都是自動ACK的，如果要手動ACK，需要改動我們的代碼：

public class Recv2 {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 創建通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
                // 手動進行ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 監聽隊列，第二個參數false，手動進行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

　　

注意到最后一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

 

如果第二個參數為true，則會自動進行ACK；如果為false，則需要手動ACK。方法的聲明：

 

 

work消息模型

說明

在剛才的基本模型中，一個生產者，一個消費者，生產的消息直接被消費者消費。比較簡單。

Work queues，也被稱為（Task queues），任務模型。

當消息處理比較耗時的時候，可能生產消息的速度會遠遠大於消息的消費速度。長此以往，消息就會堆積越來越多，無法及時處理。此時就可以使用work 模型：讓多個消費者綁定到一個隊列，共同消費隊列中的消息。隊列中的消息一旦消費，就會消失，因此任務是不會被重復執行的。

 

角色：

• P：生產者：任務的發布者

• C1：消費者，領取任務並且完成任務，假設完成速度較慢

• C2：消費者2：領取任務並完成任務，假設完成速度快

生產者

生產者與案例1中的幾乎一樣：

public class Send {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循環發布任務
        for (int i = 0; i < 50; i++) {
            // 消息內容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 關閉通道和連接
        channel.close();
        connection.close();
    }
}

　　

不過這里我們是循環發送50條消息。

消費者1

 

消費者2

 

 

與消費者1基本類似，就是沒有設置消費耗時時間。

這里是模擬有些消費者快，有些比較慢。

 

接下來，兩個消費者一同啟動，然后發送50條消息：

 

 

可以發現，兩個消費者各自消費了25條消息，而且各不相同，這就實現了任務的分發。

 

能者多勞

剛才的實現有問題嗎？

• 消費者1比消費者2的效率要低，一次任務的耗時較長

• 然而兩人最終消費的消息數量是一樣的

• 消費者2大量時間處於空閑狀態，消費者1一直忙碌

現在的狀態屬於是把任務平均分配，正確的做法應該是消費越快的人，消費的越多。

怎么實現呢？

我們可以修改設置，讓消費者同一時間只接收一條消息，這樣處理完成之前，就不會接收更多消息，就可以讓處理快的人，接收更多消息 ：

 

 

再次測試：

 

訂閱模型分類

訂閱模型示意圖：

 

前面2個案例中，只有3個角色：

• P：生產者，也就是要發送消息的程序

• C：消費者：消息的接受者，會一直等待消息到來。

• queue：消息隊列，圖中紅色部分。類似一個郵箱，可以緩存消息；生產者向其中投遞消息，消費者從其中取出消息。

而在訂閱模型中，多了一個exchange角色，而且過程略有變化：

• P：生產者，也就是要發送消息的程序，但是不再發送到隊列中，而是發給X（交換機）

• C：消費者，消息的接受者，會一直等待消息到來。

• Queue：消息隊列，接收消息、緩存消息。

• Exchange：交換機，圖中的X。一方面，接收生產者發送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。Exchange有以下3種類型：

  • Fanout：廣播，將消息交給所有綁定到交換機的隊列

  • Direct：定向，把消息交給符合指定routing key 的隊列

  • Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列

Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那么消息會丟失！

 

訂閱模型-Fanout

Fanout，也稱為廣播。

流程說明

流程圖：

 

在廣播模式下，消息發送流程是這樣的：

• 1） 可以有多個消費者

• 2） 每個消費者有自己的queue（隊列）

• 3） 每個隊列都要綁定到Exchange（交換機）

• 4） 生產者發送的消息，只能發送到交換機，交換機來決定要發給哪個隊列，生產者無法決定。

• 5） 交換機把消息發送給綁定過的所有隊列

• 6） 隊列的消費者都能拿到消息。實現一條消息被多個消費者消費

生產者

兩個變化：

• 1） 聲明Exchange，不再聲明Queue

• 2） 發送消息到Exchange，不再發送到Queue

public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        
        // 聲明exchange，指定類型為fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息內容
        String message = "Hello everyone";
        // 發布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生產者] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

　　

消費者1

public class Recv {
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 綁定隊列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

要注意代碼中：隊列需要和交換機綁定

消費者2

public class Recv2 {
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 綁定隊列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，手動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

測試

我們運行兩個消費者，然后發送1條消息：

 

 

訂閱模型-Direct

說明

在Fanout模式中，一條消息，會被所有訂閱的隊列都消費。但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。

在Direct模型下：

• 隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）

• 消息的發送方在 向 Exchange發送消息時，也必須指定消息的 RoutingKey。

• Exchange不再把消息交給每一個綁定的隊列，而是根據消息的Routing Key進行判斷，只有隊列的Routingkey與消息的 Routing key完全一致，才會接收到消息

流程圖：

 

圖解：

• P：生產者，向Exchange發送消息，發送消息時，會指定一個routing key。

• X：Exchange（交換機），接收生產者的消息，然后把消息遞交給 與routing key完全匹配的隊列

• C1：消費者，其所在隊列指定了需要routing key 為 error 的消息

• C2：消費者，其所在隊列指定了需要routing key 為 info、error、warning 的消息

生產者

此處我們模擬商品的增刪改，發送消息的RoutingKey分別是：insert、update、delete

public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明exchange，指定類型為direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息內容
        String message = "商品新增了， id = 1001";
        // 發送消息，並且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服務：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

　　

消費者1

我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品。

public class Recv {
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。假設此處需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

消費者2

我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。

public class Recv2 {
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

測試

我們分別發送增、刪、改的RoutingKey，發現結果：

 

 

訂閱模型-Topic

說明

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符！

 

Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符規則：

#：匹配一個或多個詞

*：匹配不多不少恰好1個詞

 

舉例：

audit.#：能夠匹配audit.irs.corporate 或者 audit.irs

audit.*：只能匹配audit.irs

​

圖示：

 

解釋：

• 紅色Queue：綁定的是usa.# ，因此凡是以 usa.開頭的routing key 都會被匹配到

• 黃色Queue：綁定的是#.news ，因此凡是以 .news結尾的 routing key 都會被匹配

生產者

使用topic類型的Exchange，發送消息的routing key有3種： item.isnert、item.update、item.delete：

public class Send {
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明exchange，指定類型為topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息內容
        String message = "新增商品 : id = 1001";
        // 發送消息，並且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
        System.out.println(" [商品服務：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

　　

消費者1

我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品

public class Recv {
    private final static String QUEUE_NAME = "topic_exchange_queue_1";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。需要 update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

消費者2

我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。

　　

/**
 * 消費者2
 */
public class Recv2 {
    private final static String QUEUE_NAME = "topic_exchange_queue_2";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

　　

 

---

 

 

持久化

如何避免消息丟失？

1） 消費者的ACK機制。可以防止消費者丟失消息。

2） 但是，如果在消費者消費之前，MQ就宕機了，消息就沒了。

 

是可以將消息進行持久化呢？

要將消息持久化，前提是：隊列、Exchange都持久化

交換機持久化

隊列持久化

消息持久化

 

解決消息丟失？

• ack（消費者確認）

• 持久化

• 發送消息前，將消息持久化到數據庫，並記錄消息狀態（可靠消息服務）

• 生產者確認（publisher confirm）

 

---

Spring AMQP

簡介

Sprin有很多不同的項目，其中就有對AMQP的支持：

Spring AMQP的頁面：http://projects.spring.io/spring-amqp/

注意這里一段描述：

     Spring-amqp是對AMQP協議的抽象實現，而spring-rabbit 是對協議的具體實現，也是目前的唯一實現。底層使用的就是RabbitMQ。

 

 

依賴和配置

添加AMQP的啟動器：

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

 

在application.yml中添加RabbitMQ地址：

 

spring:

  rabbitmq:

    host: 192.168.56.101

    username: leyou

    password: leyou

    virtual-host: /leyou

 

監聽者

在SpringAmqp中，對消息的消費者進行了封裝和抽象，一個普通的JavaBean中的普通方法，只要通過簡單的注解，就可以成為一個消費者。

@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

　　

• @Componet：類上的注解，注冊到Spring容器

• @RabbitListener：方法上的注解，聲明這個方法是一個消費者方法，需要指定下面的屬性：

  • bindings：指定綁定關系，可以有多個。值是@QueueBinding的數組。@QueueBinding包含下面屬性：

    • value：這個消費者關聯的隊列。值是@Queue，代表一個隊列

    • exchange：隊列所綁定的交換機，值是@Exchange類型

    • key：隊列和交換機綁定的RoutingKey

類似listen這樣的方法在一個類中可以寫多個，就代表多個消費者。

 

AmqpTemplate

Spring最擅長的事情就是封裝，把他人的框架進行封裝和整合。

Spring為AMQP提供了統一的消息處理模板：AmqpTemplate，非常方便的發送消息，其發送方法：

紅框圈起來的是比較常用的3個方法，分別是：

• 指定交換機、RoutingKey和消息體

• 指定消息

• 指定RoutingKey和消息，會向默認的交換機發送消息

測試代碼

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
        // 等待10秒后再結束
        Thread.sleep(10000);
    }
}

　　運行后查看日志：