RabbitMQ （二）工作隊列

jopen 8年前發布 | 9K 次閱讀消息系統

轉載請標明出處：http://blog.csdn.net/lmj623565791/article/details/37620057

本系列教程主要來自于官網入門教程的翻譯，然后自己進行了部分的修改與實驗，內容僅供參考。

上一篇博客中我們寫了通過一個命名的隊列發送和接收消息，如果你還不了解請點擊：RabbitMQ 入門 Helloworld。這篇中我們將會創建一個工作隊列用來在工作者（consumer）間分發耗時任務。

工作隊列的主要任務是：避免立刻執行資源密集型任務，然后必須等待其完成。相反地，我們進行任務調度：我們把任務封裝為消息發送給隊列。工作進行在后臺運行并不斷的從隊列中取出任務然后執行。當你運行了多個工作進程時，任務隊列中的任務將會被工作進程共享執行。
這樣的概念在web應用中極其有用，當在很短的HTTP請求間需要執行復雜的任務。

1、準備

我們使用Thread.sleep來模擬耗時的任務。我們在發送到隊列的消息的末尾添加一定數量的點，每個點代表在工作線程中需要耗時1秒，例如hello…將會需要等待3秒。

發送端：

NewTask.java

package com.zhy.rabbit._02_workqueue;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class NewTask
{
    //隊列名稱
    private final static String QUEUE_NAME = "workqueue";

    public static void main(String[] args) throws IOException
    {
        //創建連接和頻道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //發送10條消息，依次在消息后面附加1-10個點
        for (int i = 0; i < 10; i++)
        {
            String dots = "";
            for (int j = 0; j <= i; j++)
            {
                dots += ".";
            }
            String message = "helloworld" + dots+dots.length();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
        //關閉頻道和資源
        channel.close();
        connection.close();

    }


}

接收端：

Work.java

package com.zhy.rabbit._02_workqueue;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
    //隊列名稱
    private final static String QUEUE_NAME = "workqueue";

    public static void main(String[] argv) throws java.io.IOException,
            java.lang.InterruptedException
    {
        //區分不同工作進程的輸出
        int hashCode = Work.class.hashCode();
        //創建連接和頻道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(hashCode
                + " [*] Waiting for messages. To exit press CTRL+C");

        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 指定消費隊列
        channel.basicConsume(QUEUE_NAME, true, consumer);
        while (true)
        {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());

            System.out.println(hashCode + " [x] Received '" + message + "'");
            doWork(message);
            System.out.println(hashCode + " [x] Done");

        }

    }

    /**
     * 每個點耗時1s
     * @param task
     * @throws InterruptedException
     */
    private static void doWork(String task) throws InterruptedException
    {
        for (char ch : task.toCharArray())
        {
            if (ch == '.')
                Thread.sleep(1000);
        }
    }
}

Round-robin 轉發
使用任務隊列的好處是能夠很容易的并行工作。如果我們積壓了很多工作，我們僅僅通過增加更多的工作者就可以解決問題，使系統的伸縮性更加容易。
下面我們先運行3個工作者（Work.java）實例，然后運行NewTask.java，3個工作者實例都會得到信息。但是如何分配呢？讓我們來看輸出結果：[x] Sent 'helloworld.1'
[x] Sent 'helloworld..2'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld.....5'
[x] Sent 'helloworld......6'
[x] Sent 'helloworld.......7'
[x] Sent 'helloworld........8'
[x] Sent 'helloworld.........9'
[x] Sent 'helloworld..........10'

工作者1：
605645 [*] Waiting for messages. To exit press CTRL+C
605645 [x] Received 'helloworld.1'
605645 [x] Done
605645 [x] Received 'helloworld....4'
605645 [x] Done
605645 [x] Received 'helloworld.......7'
605645 [x] Done
605645 [x] Received 'helloworld..........10'
605645 [x] Done

工作者2：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld..2'
18019860 [x] Done
18019860 [x] Received 'helloworld.....5'
18019860 [x] Done
18019860 [x] Received 'helloworld........8'
18019860 [x] Done

工作者3：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld...3'
18019860 [x] Done
18019860 [x] Received 'helloworld......6'
18019860 [x] Done
18019860 [x] Received 'helloworld.........9'
18019860 [x] Done
可以看到，默認的，RabbitMQ會一個一個的發送信息給下一個消費者(consumer)，而不考慮每個任務的時長等等，且是一次性分配，并非一個一個分配。平均的每個消費者將會獲得相等數量的消息。這樣分發消息的方式叫做round-robin。

2、消息應答（message acknowledgments）

執行一個任務需要花費幾秒鐘。你可能會擔心當一個工作者在執行任務時發生中斷。我們上面的代碼，一旦RabbItMQ交付了一個信息給消費者，會馬上從內存中移除這個信息。在這種情況下，如果殺死正在執行任務的某個工作者，我們會丟失它正在處理的信息。我們也會丟失已經轉發給這個工作者且它還未執行的消息。
上面的例子，我們首先開啟兩個任務，然后執行發送任務的代碼（NewTask.java），然后立即關閉第二個任務，結果為：
工作者2：

31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld..2'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'

工作者1：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld.1'
18019860 [x] Done
18019860 [x] Received 'helloworld...3'
18019860 [x] Done
18019860 [x] Received 'helloworld.....5'
18019860 [x] Done
18019860 [x] Received 'helloworld.......7'
18019860 [x] Done
18019860 [x] Received 'helloworld.........9'
18019860 [x] Done
可以看到，第二個工作者至少丟失了6，8，10號任務，且4號任務未完成。

但是，我們不希望丟失任何任務（信息）。當某個工作者（接收者）被殺死時，我們希望將任務傳遞給另一個工作者。
為了保證消息永遠不會丟失，RabbitMQ支持消息應答（message acknowledgments）。消費者發送應答給RabbitMQ，告訴它信息已經被接收和處理，然后RabbitMQ可以自由的進行信息刪除。
如果消費者被殺死而沒有發送應答，RabbitMQ會認為該信息沒有被完全的處理，然后將會重新轉發給別的消費者。通過這種方式，你可以確認信息不會被丟失，即使消者偶爾被殺死。
這種機制并沒有超時時間這么一說，RabbitMQ只有在消費者連接斷開是重新轉發此信息。如果消費者處理一個信息需要耗費特別特別長的時間是允許的。
消息應答默認是打開的。上面的代碼中我們通過顯示的設置autoAsk=true關閉了這種機制。下面我們修改代碼（Work.java）：

boolean ack = false ; //打開應答機制
channel.basicConsume(QUEUE_NAME, ack, consumer);
//另外需要在每次處理完成一個消息后，手動發送一次應答。
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

完整修改后的Work.java

package com.zhy.rabbit._02_workqueue.ack;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
    //隊列名稱
    private final static String QUEUE_NAME = "workqueue";

    public static void main(String[] argv) throws java.io.IOException,
            java.lang.InterruptedException
    {
        //區分不同工作進程的輸出
        int hashCode = Work.class.hashCode();
        //創建連接和頻道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(hashCode
                + " [*] Waiting for messages. To exit press CTRL+C");
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 指定消費隊列
        boolean ack = false ; //打開應答機制
        channel.basicConsume(QUEUE_NAME, ack, consumer);
        while (true)
        {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());

            System.out.println(hashCode + " [x] Received '" + message + "'");
            doWork(message);
            System.out.println(hashCode + " [x] Done");
            //發送應答
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        }

    }
}

測試：
我們把消息數量改為5，然后先打開兩個消費者（Work.java），然后發送任務（NewTask.java），立即關閉一個消費者，觀察輸出：
[x] Sent 'helloworld.1'
[x] Sent 'helloworld..2'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld.....5'

工作者2
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld..2'
18019860 [x] Done
18019860 [x] Received 'helloworld....4'

工作者1
31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld.1'
31054905 [x] Done
31054905 [x] Received 'helloworld...3'
31054905 [x] Done
31054905 [x] Received 'helloworld.....5'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'
31054905 [x] Done

可以看到工作者2沒有完成的任務4，重新轉發給工作者1進行完成了。

3、消息持久化（Message durability）

我們已經學習了即使消費者被殺死，消息也不會被丟失。但是如果此時RabbitMQ服務被停止，我們的消息仍然會丟失。

當RabbitMQ退出或者異常退出，將會丟失所有的隊列和信息，除非你告訴它不要丟失。我們需要做兩件事來確保信息不會被丟失：我們需要給所有的隊列和消息設置持久化的標志。
第一，我們需要確認RabbitMQ永遠不會丟失我們的隊列。為了這樣，我們需要聲明它為持久化的。
boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
注：RabbitMQ不允許使用不同的參數重新定義一個隊列，所以已經存在的隊列，我們無法修改其屬性。
第二，我們需要標識我們的信息為持久化的。通過設置MessageProperties（implements BasicProperties）值為PERSISTENT_TEXT_PLAIN。
channel.basicPublish("", "task_queue",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
現在你可以執行一個發送消息的程序，然后關閉服務，再重新啟動服務，運行消費者程序做下實驗。

4、公平轉發（Fair dispatch）

或許會發現，目前的消息轉發機制（Round-robin）并非是我們想要的。例如，這樣一種情況，對于兩個消費者，有一系列的任務，奇數任務特別耗時，而偶數任務卻很輕松，這樣造成一個消費者一直繁忙，另一個消費者卻很快執行完任務后等待。
造成這樣的原因是因為RabbitMQ僅僅是當消息到達隊列進行轉發消息。并不在乎有多少任務消費者并未傳遞一個應答給RabbitMQ。僅僅盲目轉發所有的奇數給一個消費者，偶數給另一個消費者。
為了解決這樣的問題，我們可以使用basicQos方法，傳遞參數為prefetchCount = 1。這樣告訴RabbitMQ不要在同一時間給一個消費者超過一條消息。換句話說，只有在消費者空閑的時候會發送下一條信息。

int prefetchCount = 1;
channel.basicQos(prefetchCount);

注：如果所有的工作者都處于繁忙狀態，你的隊列有可能被填充滿。你可能會觀察隊列的使用情況，然后增加工作者，或者使用別的什么策略。
測試：改變發送消息的代碼，將消息末尾點數改為6-2個，然后首先開啟兩個工作者，接著發送消息：

[x] Sent 'helloworld......6'
[x] Sent 'helloworld.....5'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld..2'

工作者1：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld......6'
18019860 [x] Done
18019860 [x] Received 'helloworld...3'
18019860 [x] Done

工作者2：
31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld.....5'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'
31054905 [x] Done
31054905 [x] Received 'helloworld..2'
31054905 [x] Done

可以看出此時并沒有按照之前的Round-robin機制進行轉發消息，而是當消費者不忙時進行轉發。且這種模式下支持動態增加消費者，因為消息并沒有發送出去，動態增加了消費者馬上投入工作。而默認的轉發機制會造成，即使動態增加了消費者，此時的消息已經分配完畢，無法立即加入工作，即使有很多未完成的任務。

5、完整的代碼

NewTask.java

package com.zhy.rabbit._02_workqueue.ackandpersistence;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

public class NewTask
{
    // 隊列名稱
    private final static String QUEUE_NAME = "workqueue_persistence";

    public static void main(String[] args) throws IOException
    {
        // 創建連接和頻道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 聲明隊列
        boolean durable = true;// 1、設置隊列持久化
        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
        // 發送10條消息，依次在消息后面附加1-10個點
        for (int i = 5; i > 0; i--)
        {
            String dots = "";
            for (int j = 0; j <= i; j++)
            {
                dots += ".";
            }
            String message = "helloworld" + dots + dots.length();
            // MessageProperties 2、設置消息持久化
            channel.basicPublish("", QUEUE_NAME,
                    MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
        // 關閉頻道和資源
        channel.close();
        connection.close();

    }

}

Work.java

package com.zhy.rabbit._02_workqueue.ackandpersistence;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
    // 隊列名稱
    private final static String QUEUE_NAME = "workqueue_persistence";

    public static void main(String[] argv) throws java.io.IOException,
            java.lang.InterruptedException
    {
        // 區分不同工作進程的輸出
        int hashCode = Work.class.hashCode();
        // 創建連接和頻道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 聲明隊列
        boolean durable = true;
        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
        System.out.println(hashCode
                + " [*] Waiting for messages. To exit press CTRL+C");
        //設置最大服務轉發消息數量
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 指定消費隊列
        boolean ack = false; // 打開應答機制
        channel.basicConsume(QUEUE_NAME, ack, consumer);
        while (true)
        {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());

            System.out.println(hashCode + " [x] Received '" + message + "'");
            doWork(message);
            System.out.println(hashCode + " [x] Done");
            //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        }

    }

    /**
     * 每個點耗時1s
     * 
     * @param task
     * @throws InterruptedException
     */
    private static void doWork(String task) throws InterruptedException
    {
        for (char ch : task.toCharArray())
        {
            if (ch == '.')
                Thread.sleep(1000);
        }
    }
}

來自： http://blog.csdn.net//lmj623565791/article/details/37620057

本文由用戶 jopen 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1452954658308.html

消息系統

RabbitMQ （二）工作隊列

1、準備

2、消息應答（message acknowledgments）

3、消息持久化（Message durability）

4、公平轉發（Fair dispatch）

5、完整的代碼

相關經驗

相關資訊

相關文檔

目錄

RabbitMQ （二）工作隊列

1、 準備

2、 消息應答（message acknowledgments）

3、 消息持久化（Message durability）

4、公平轉發（Fair dispatch）

5、完整的代碼

相關經驗

相關資訊

相關文檔

目錄

1、準備

2、消息應答（message acknowledgments）

3、消息持久化（Message durability）