在上篇中我們實現(xiàn)了程序來從一個已經(jīng)命名的隊列里發(fā)送和接收消息。本篇博文中我們將要創(chuàng)建工作隊列用來把一些比較耗時的任務(wù)分配給多個worker揽乱。
工作隊列的主要思想就是避開立刻處理某個資源消耗交大的任務(wù)并且需要等待它執(zhí)行完成晦闰。取而代之的是我們可以將它加入計劃列表朝蜘，并在后邊執(zhí)行這些任務(wù)蚂四。我們將任務(wù)分裝成一個消息歼疮，并發(fā)送到隊列中署咽。后臺的工作程序在接收到消息后將會立刻執(zhí)行任務(wù)近顷。當運行多個執(zhí)行器時，任務(wù)將會在他們之間共享宁否。
為什么會出現(xiàn)work queues?
前提:使用simple隊列的時候 我們應(yīng)用程序在是使用消息系統(tǒng)的時候,一般生產(chǎn)者P生產(chǎn)消息是毫不費力的(發(fā)送消息即可),而消費者接收完消息后的需要處理,會耗費一定的時間,這時候,就有可能導致很多消息堆積在隊列里面,一個消費者有可能不夠用 那么怎么讓消費者同事處理多個消息呢? 在同一個隊列上創(chuàng)建多個消費者,讓他們相互競爭,這樣消費者就可以同時處理多條消息了 使用任務(wù)隊列的優(yōu)點之一就是可以輕易的并行工作窒升。如果我們積壓了好多工作，我們可以通過增加工作者（消費者）來解決這一問題慕匠，使得系統(tǒng)的伸縮性更加容易饱须。
1.Round-robin（輪詢分發(fā)）
輪詢分發(fā)結(jié)果就是不管服務(wù)器誰忙或清閑，都不會給誰多一個任務(wù)或少一個任務(wù)台谊，任務(wù)總是你一個我一個的分蓉媳。
1.1 消息提供者
package com.hrabbit.rabbitmq.send;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:37
* @Description:
*/
public class Send {
private final static String QUEUE_NAME = "hrabbit_queue_work";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
// 獲取到連接以及mq通道
Connection connection = ConnectionUtils.getConnection();
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 0; i < 50; i++) {
// 消息內(nèi)容
String message = "." + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(i * 10);
}
channel.close();
connection.close();
}
}
1.1 消費者一號
package com.hrabbit.rabbitmq.recover;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:42
* @Description:
*/
public class Received {
private final static String QUEUE_NAME = "hrabbit_queue_work";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
// 聲明隊列譬挚，主要為了防止消息接收者先運行此程序，隊列還不存在時創(chuàng)建隊列酪呻。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//定義一個消息的消費者
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [1] 消費消息: '" + message + "'");
try {
doWork(message);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(" [1] 消費消息結(jié)束");
}
}
};
boolean autoAck = true;
//消息的確認模式自動應(yīng)答
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
private static void doWork(String task) throws InterruptedException {
Thread.sleep(1000);
}
}
1.1 消費者二號
package com.hrabbit.rabbitmq.recover;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:42
* @Description:
*/
public class Received2 {
private final static String QUEUE_NAME = "hrabbit_queue_work";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
// 聲明隊列减宣，主要為了防止消息接收者先運行此程序，隊列還不存在時創(chuàng)建隊列玩荠。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//定義一個消息的消費者
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [2] 消費消息: '" + message + "'");
try {
doWork(message);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(" [2] 消費消息結(jié)束");
}
}
};
boolean autoAck = true;
//消息的確認模式自動應(yīng)答
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
private static void doWork(String task) throws InterruptedException {
Thread.sleep(2000);
}
}
最后消費者一號和消費者二號的輸出結(jié)果：
消費者一號：
消費者二號：
注:消費者1 我們處理時間是1s ;而消費者2中處理時間是2s; 但是我們看到的現(xiàn)象并不是 1處理的多 消費者2處理的少蚪腋，消費者1中將偶數(shù)部分處理掉了 ，消費者2中將基數(shù)部分處理掉了 姨蟋。
測試結(jié)果:
1.消費者1和消費者2獲取到的消息內(nèi)容是不同的,同一個消息只能被一個消費者獲取
2.消費者1和消費者2貨到的消息數(shù)量是一樣的 一個奇數(shù)一個偶數(shù) 按道理消費者1 獲取的比消費者2要多
這種方式叫做輪詢分發(fā) 結(jié)果就是不管誰忙或清閑屉凯，都不會給誰多一個任務(wù)或少一個任務(wù)，任務(wù)總是你一個我一個的分眼溶。
2.Fair dispatch（公平分發(fā)）
雖然上面的分配法方式也還行悠砚，但是有個問題就是：比如：現(xiàn)在有2個消費者，所有的偶數(shù)的消息都是繁忙的堂飞，而奇數(shù)則是輕松的灌旧。按照輪詢的方式，偶數(shù)的任務(wù)交給了第一個消費者绰筛，所以一直在忙個不停枢泰。奇數(shù)的任務(wù)交給另一個消費者，則立即完成任務(wù)铝噩，然后閑得不行衡蚂。 而RabbitMQ則是不了解這些的。他是不知道你消費者的消費能力的,這是因為當消息進入隊列骏庸，RabbitMQ就會分派消息毛甲。而rabbitmq只是盲目的將消息輪詢的發(fā)給消費者。你一個我一個的這樣發(fā)送.
為了解決這個問題具被，我們使用basicQos( prefetchCount = 1)方法玻募，來限制RabbitMQ只發(fā)不超過1條的消息給同一個消費者。當消息處理完畢后一姿，有了反饋ack七咧，才會進行第二次發(fā)送。(也就是說需要手動反饋給Rabbitmq ) 還有一點需要注意叮叹，使用公平分發(fā)艾栋，必須關(guān)閉自動應(yīng)答，改為手動應(yīng)答衬横。
2.1 生產(chǎn)者
代碼基本不用修改裹粤，只需要添加一行代碼：channel.basicQos(1);
package com.hrabbit.rabbitmq.send;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:37
* @Description:
*/
public class Send {
private final static String QUEUE_NAME = "hrabbit_queue_work";
private static Integer prefetchCount=1;
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
// 獲取到連接以及mq通道
Connection connection = ConnectionUtils.getConnection();
Channel channel = connection.createChannel();
//每個消費者發(fā)送確認信號之前，消息隊列不發(fā)送下一個消息過來，一次只處理一個消息
//限制發(fā)給同一個消費者不得超過1條消息
channel.basicQos(prefetchCount);
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 0; i < 50; i++) {
// 消息內(nèi)容
String message = "." + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(i * 10);
}
channel.close();
connection.close();
}
}
2.2 消費者一
首先設(shè)置channel.basicQos(1);保證一次只分發(fā)一次遥诉，其次將autoAck設(shè)置為fase,在每次消息處理完成之后拇泣，手動確認消息channel.basicAck(envelope.getDeliveryTag(), false);
package com.hrabbit.rabbitmq.recover;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:42
* @Description:
*/
public class Received {
private final static String QUEUE_NAME = "hrabbit_queue_work";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
// 聲明隊列，主要為了防止消息接收者先運行此程序矮锈，隊列還不存在時創(chuàng)建隊列霉翔。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);//保證一次只分發(fā)一個
//定義一個消息的消費者
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [1] 消費消息: '" + message + "'");
try {
doWork(message);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(" [1] 消費消息結(jié)束");
//手動應(yīng)答
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
//消息的確認模式關(guān)閉自動
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
private static void doWork(String task) throws InterruptedException {
Thread.sleep(1000);
}
}
2.2 消費者二
package com.hrabbit.rabbitmq.recover;
import com.hrabbit.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: hrabbit
* @Date: 2018-06-27 下午2:42
* @Description:
*/
public class Received2 {
private final static String QUEUE_NAME = "hrabbit_queue_work";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
// 聲明隊列，主要為了防止消息接收者先運行此程序苞笨，隊列還不存在時創(chuàng)建隊列债朵。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);//保證一次只分發(fā)一個
//定義一個消息的消費者
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [2] 消費消息: '" + message + "'");
try {
doWork(message);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(" [2] 消費消息結(jié)束");
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
//消息的確認模式自動應(yīng)答關(guān)閉
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
private static void doWork(String task) throws InterruptedException {
//消息的處理時間為2000
Thread.sleep(2000);
}
}
最后消費者一號和消費者二號的輸出結(jié)果：
消費者一：
消費者二：
這時候現(xiàn)象就是消費者1 速度大于消費者2 。
系列文章：
RabbitMQ：RabbitMQ-理論基礎(chǔ)
RabbitMQ：快速入門hello word
RabbitMQ：RabbitMQ：消息應(yīng)答與消息持久化
RabbitMQ：發(fā)布/訂閱 Publish/Subscribe
RabbitMQ：路由Routing
RabbitMQ：Topic類型的exchange
RabbitMQ：RabbitMQ之消息確認機制（事務(wù)+Confirm）
RabbitMQ：spring整合RabbitMQ
