在上一篇文章中我们解决了最简单的helloworld 消息传递，这一篇中我们来探讨rabbitMQ中的任务分发
rabbitMQ任务分发机制的核心出发点就是避免立刻进行“资源密集”或者说time-consuming的任务，因为这样就必须同步等待耗时任务的完成。取而代之的是schedule这些任务再稍后完成，在本篇的demo中我们将task封装成一条message将其发送到队列中。一个后台运行的worker进程会从队列中获取message并执行任务。
任务分发机制在web应用中非常有用，因为通常我们不会在一次http请求响应过程中处理复杂的耗时任务。 当有Consumer需要大量的运算时，RabbitMQ Server需要一定的分发机制来balance每个Consumer的load。
rabbitMQ的任务分发机制模型如下图所示：
准备
在上一篇文章中的实例中，我们发送一个“hello world”的消息，在这篇文章中，我们发送一个字符串代表复杂的任务，用thread.sleep()函数模拟可能的操作，比如图片的resize，pdf的内容渲染或者提取。
复用上文中的code，为了便于区别，我们还是命名为new_task.java
String[] messages = {"a","b","c","d"};
String message = getMessage(messages);
//the concept of channel in rabbitMQ,the first parameter defines the name of exchange,
//"" means the default exchange
channel.basicPublish("", TASK_QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
getMessage方法，非常简单：
private static String getMessage(String[] strings) {
if (strings.length < 1)
return "Hello World!";
return joinStrings(strings, ".");
}
private static String joinStrings(String[] strings, String delimiter) {
int length = strings.length;
if (length == 0) return "";
StringBuilder words = new StringBuilder(strings[0]);
for (int i = 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
原来的receiver代码也需要略作改动，同样为了便于理解，我们将其重新命名为worker.java,并且根据message中的“.”进行任务处理的模拟。
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
// TODO Auto-generated method stub
//super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
try {
doWork(message);
doWork方法：
private static void doWork(String task){
for(char c : task.toCharArray()){
System.out.print(c + "\t");
if(c == '.'){
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
Thread.currentThread().interrupt();
}
}
}
}
至此我们已经完成了工作的大半。
round-robin dispatching 循环分发
RabbitMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的。如果现在load加重，那么只需要创建更多的Consumer来进行任务处理即可。首先我们来运行两个worker实例，这里通过命令行的方式完成：
shell1$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarWorker
[*] Waiting for messages. To exit press CTRL+C
shell2$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarWorker
[*] Waiting for messages. To exit press CTRL+C
然后producer将要发布新任务：
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarNewTask First message.
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarNewTask Second message..
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarNewTask Third message...
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarNewTask Fourth message....
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jarNewTask Fifth message.....
我们来观察一下 worker收到的消息
shell2$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
Worker
[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Second message..'
[x] Received 'Fourth message....'
默认情况下，rabbitMQ会按顺序的将message依次分发给下一个consumer，这种分发方式就叫做round-robin。
Message Acknowledgement 消息确认
运行一个任务可能需要好几秒甚至更久，那么有个问题值得探究，如果一个consumer开始了一段长任务，但是在任务处理到一半时consumer进程异常退出会发生什么。不幸的是，如果我们采用no-ack的方式，这个消息就消失了。也就是说，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。
如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。 为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。
在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。 如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。 这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。
message ack 默认情况下是开启的，在上一节中我们通过autoAck=true来显式的关闭了acknowledgement，现在我们修改handleDelivery回调函数，来发送确认信息。
channel.basicQos(1);
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(" [x] Received '" + message + "'");
try {
doWork(message);
} finally {
System.out.println(" [x] Done");
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
Message Durability 消息持久化
上文中我们学习了在consumer异常退出或者中断的情况下如何通过消息确认来保证消息的不丢失，但是在rabbitMQ server异常退出或者中断情况下就无能为力了，这种情况持久化消息可以帮忙。消息持久化需要做两件事情就是声明queue和message都是durable的：
boolean durable = true;channel.queueDeclare("hello", durable, false, false, null);
上述语句执行不会有什么错误，但是确得不到我们想要的结果，原因就是RabbitMQ Server已经维护了一个叫hello的queue，那么上述执行不会有任何的作用，也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。那么workaround也很简单，声明一个另外的名字的queue，比如名字定位task_queue：
boolean durable = true;channel.queueDeclare("task_queue", durable, false, false, null);
再次强调，Producer和Consumer都应该去创建这个queue，尽管只有一个地方的创建是真正起作用的。接下来，需要持久化Message，即在Publish的时候指定一个properties，方式如下：
import com.rabbitmq.client.MessageProperties;
channel.basicPublish("", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
Fair Dispatch 公平分发
你可能也注意到了，分发机制不是那么优雅。默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message，只是按照这个默认机制进行分发。 那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却是毫无休息的机会。那么，RabbitMQ是如何处理这种问题呢？
过 basic.qos 方法设置prefetch_count=1 。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。 设置方法如下：
int prefetchCount = 1;channel.basicQos(prefetchCount);
整合后的整个代码如下：
new task.java
package cn.edu.nju.liushao.worker;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
public class NewTask {
private static final String TASK_QUEUE_NAME = "task_queue";
private static final String MQ_ADDRESS = "localhost";
public static void main(String[] argv) throws Exception {
/*
* init factory,connection and channel
*/
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(MQ_ADDRESS);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//declare a queue
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
String[] messages = {"a","b","c","d"};
String message = getMessage(messages);
//the concept of channel in rabbitMQ,the first parameter defines the name of exchange,
//"" means the default exchange
channel.basicPublish("", TASK_QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
private static String getMessage(String[] strings) {
if (strings.length < 1)
return "Hello World!";
return joinStrings(strings, ".");
}
private static String joinStrings(String[] strings, String delimiter) {
int length = strings.length;
if (length == 0) return "";
StringBuilder words = new StringBuilder(strings[0]);
for (int i = 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
worker.java
package cn.edu.nju.liushao.worker;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
public class Worker {
private static final String TASK_QUEUE_NAME = "task_queue";
private static final String MQ_ADDRESS = "localhost";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(MQ_ADDRESS);
final Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
System.out.println(" [*] waiting for messages. To exit press CTRL+C");
channel.basicQos(1);
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
// TODO Auto-generated method stub
//super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
try {
doWork(message);
} finally {
System.out.println("[x] done");
// send back acknowledgement
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
channel.basicConsume(TASK_QUEUE_NAME, false,consumer);
}
private static void doWork(String task){
for(char c : task.toCharArray()){
System.out.print(c + "\t");
if(c == '.'){
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
Thread.currentThread().interrupt();
}
}
}
}
}
