前言
RabbitMQ六種隊列模式-簡單隊列
RabbitMQ六種隊列模式-工作隊列 [本文]
RabbitMQ六種隊列模式-釋出訂閱
RabbitMQ六種隊列模式-路由模式
RabbitMQ六種隊列模式-主題模式
上文我們了解了 RabbitMQ 六種隊列模式中的簡單隊列，代碼也是非常的簡單，比較容易了解。
但是簡單隊列有個缺點，簡單隊列是一一對應的關系，即點對點，一個生産者對應一個消費者，按照這個邏輯，如果我們有一些比較耗時的任務，也就意味着需要大量的時間才能處理完畢，顯然簡單隊列模式并不能滿足我們的工作需求，我們今天再來看看工作隊列。
文章目錄
1. 什麼是工作隊列2. 代碼部分2.1 生産者2.2 消費者3. 循環分發3.1 啟動生産者3.2 啟動兩個消費者3.3 公平分發4. 消息持久化4.1 問題背景4.2 參數配置5. 工作隊列總結
1. 什麼是工作隊列
工作隊列：用來将耗時的任務分發給多個消費者（工作者）
主要解決問題：處理資源密集型任務，并且還要等他完成。有了工作隊列，我們就可以将具體的工作放到後面去做，将工作封裝為一個消息，發送到隊列中，一個工作程序就可以取出消息并完成工作。如果啟動了多個工作程序，那麼工作就可以在多個程序間共享。
工作隊列也稱為公平性隊列模式，怎麼個說法呢？
循環分發，假如我們擁有兩個消費者，預設情況下，RabbitMQ 将按順序将每條消息發送給下一個消費者，平均而言，每個消費者将獲得相同數量的消息，這種分發消息的方式稱為輪詢。
看代碼吧。
2. 代碼部分
2.1 生産者
建立50個消息
public class Producer2 {
/** 隊列名稱 */
private static final String QUEUE_NAME = "test_queue";
public static void main(String[] args) throws IOException, TimeoutException {
/** 1.擷取連接配接 */
Connection newConnection = MQConnectionUtils.newConnection();
/** 2.建立通道 */
Channel channel = newConnection.createChannel();
/**3.建立隊列聲明 */
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/**保證一次隻分發一次 限制發送給同一個消費者 不得超過一條消息 */
channel.basicQos(1);
for (int i = 1; i <= 50; i++) {
String msg = "生産者消息_" + i;
System.out.println("生産者發送消息:" + msg);
/**4.發送消息 */
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
}
channel.close();
newConnection.close();
}
}
2.2 消費者
public class Customer2_1 {
/**
* 隊列名稱
*/
private static final String QUEUE_NAME = "test_queue";
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("001");
/** 1.擷取連接配接 */
Connection newConnection = MQConnectionUtils.newConnection();
/** 2.擷取通道 */
final Channel channel = newConnection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/** 保證一次隻分發一次 限制發送給同一個消費者 不得超過一條消息 */
channel.basicQos(1);
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String msgString = new String(body, "UTF-8");
System.out.println("消費者擷取消息:" + msgString);
try {
Thread.sleep(1000);
} catch (Exception e) {
} finally {
/** 手動回執消息 */
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
/** 3.監聽隊列 */
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
}
}
3. 循環分發
3.1 啟動生産者
3.2 啟動兩個消費者
在生産者中我們發送了50條消息進入隊列，而上方消費者啟動圖裡很明顯的看到輪詢的效果，就是每個消費者會分到相同的隊列任務。
3.3 公平分發
由于上方模拟的是非常簡單的消息隊列的消費，假如有一些非常耗時的任務，某個消費者在緩慢地進行處理，而另一個消費者則空閑，顯然是非常消耗資源的。
再舉一個例子，一個1年的程式員，跟一個3年的程式員，配置設定相同的任務量，明顯3年的程式員處理起來更加得心應手，很快就無所事事了，但是3年的程式員拿着非常高的薪資！顯然3年的程式員應該承擔更多的責任，那怎麼辦呢？
公平分發。
其實發生上述問題的原因是 RabbitMQ 收到消息後就立即分發出去，而沒有确認各個工作者未傳回确認的消息數量，類似于TCP/UDP中的UDP，面向無連接配接。
是以我們可以使用 basicQos 方法，并将參數 prefetchCount 設為1，告訴 RabbitMQ 我每次值處理一條消息，你要等我處理完了再分給我下一個。這樣 RabbitMQ 就不會輪流分發了，而是尋找空閑的工作者進行分發。
關鍵性代碼：
/** 2.擷取通道 */
final Channel channel = newConnection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/** 保證一次隻分發一次 限制發送給同一個消費者 不得超過一條消息 */
channel.basicQos(1);
4. 消息持久化
4.1 問題背景
上邊我們提到的公平分發是由消費者收取消息時确認解決的，但是這裡面又會出現被 kill 的情況。
當有多個消費者同時收取消息，且每個消費者在接收消息的同時，還要處理其它的事情，且會消耗很長的時間。在此過程中可能會出現一些意外，比如消息接收到一半的時候，一個消費者死掉了。
這種情況要使用消息接收确認機制，可以執行上次當機的消費者沒有完成的事情。
但是在預設情況下，我們程式建立的消息隊列以及存放在隊列裡面的消息，都是非持久化的。當RabbitMQ死掉了或者重新開機了，上次建立的隊列、消息都不會儲存。
怎麼辦呢？
4.2 參數配置
參數配置一：生産者建立隊列聲明時，修改第二個參數為 true
/**3.建立隊列聲明 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
參數配置二：生産者發送消息時，修改第三個參數為MessageProperties.PERSISTENT_TEXT_PLAIN
for (int i = 1; i <= 50; i++) {
String msg = "生産者消息_" + i;
System.out.println("生産者發送消息:" + msg);
/**4.發送消息 */
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
}
5. 工作隊列總結
1、循環分發：消費者端在信道上打開消息應答機制，并確定能傳回接收消息的确認資訊，這樣可以保證消費者發生故障也不會丢失消息。
2、消息持久化：伺服器端和用戶端都要指定隊列的持久化和消息的持久化，這樣可以保證RabbitMQ重新開機，隊列和消息也不會丢失。
3、公平分發：指定消費者接收的消息個數，避免出現消息均勻推送出現的資源不合理利用的問題。
案例代碼：https://www.lanzous.com/i5ydu6d
18年專科畢業後，期間一度迷茫，最近我建立了一個公衆号用來記錄自己的成長。