消息隊列（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種。

可是其實三、四、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);
    }
}