RabbitMQ （二）工做隊列

時間 2019-12-04

原文原文鏈接

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

本系列教程主要來自於官網入門教程的翻譯，而後本身進行了部分的修改與實驗，內容僅供參考。 web

上一篇博客中咱們寫了經過一個命名的隊列發送和接收消息，若是你還不瞭解請點擊：RabbitMQ 入門 Helloworld。這篇中咱們將會建立一個工做隊列用來在工做者（consumer）間分發耗時任務。學習

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

一、準備

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

發送端： fetch

NewTask.java spa

[java] view plain copy

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();
}
}

接收端： .net

Work.java 線程

[java] view plain copy

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 blog

工做者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。

二、消息應（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）：

[java] view plain copy

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

完整修改後的Work.java

[java] view plain copy

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進行完成了。

三、消息持久化（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());
如今你能夠執行一個發送消息的程序，而後關閉服務，再從新啓動服務，運行消費者程序作下實驗。

四、公平轉發（Fair dispatch）

或許會發現，目前的消息轉發機制（Round-robin）並不是是咱們想要的。例如，這樣一種狀況，對於兩個消費者，有一系列的任務，奇數任務特別耗時，而偶數任務卻很輕鬆，這樣形成一個消費者一直繁忙，另外一個消費者卻很快執行完任務後等待。
形成這樣的緣由是由於RabbitMQ僅僅是當消息到達隊列進行轉發消息。並不在意有多少任務消費者並未傳遞一個應答給RabbitMQ。僅僅盲目轉發全部的奇數給一個消費者，偶數給另外一個消費者。
爲了解決這樣的問題，咱們可使用basicQos方法，傳遞參數爲prefetchCount = 1。這樣告訴RabbitMQ不要在同一時間給一個消費者超過一條消息。換句話說，只有在消費者空閒的時候會發送下一條信息。

[java] view plain copy

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機制進行轉發消息，而是當消費者不忙時進行轉發。且這種模式下支持動態增長消費者，由於消息並無發送出去，動態增長了消費者立刻投入工做。而默認的轉發機制會形成，即便動態增長了消費者，此時的消息已經分配完畢，沒法當即加入工做，即便有不少未完成的任務。