在上一篇文章中我们解决了最简单的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();
}
}
}
}
}