消息分发
在【RabbitMQ】 HelloWorld中我们写了发送/接收消息的程序。这次我们将创建一个Work Queue用来在多个消费者之间分配耗时任务。
Work Queues（又称为：Task Queues）的主要思想是：尽可能的减少执行资源密集型任务时的等待时间。我们将任务封装为消息并发送到队列，在后台的工作进程将弹出任务并进行作业。当你运行很多worker，任务将在他们之间共享。
这个概念在WEB应用中尤为有效，因为在一个HTTP请求进行复杂操作是不可能的。
准备
在上一节我们发送了一条包含“Hello World”的消息。现在我们将要发送代表复杂任务的字符串。我们没有真实场景的复杂任务，例如调整图片大小或呈现PDF文件，让我们假装自己很忙 - 通过Thread.sleep()。我们将根据字符串中“.”的数量来衡量任务复杂度；每一个“.”增加1秒钟的工作时间。例如：一个“Hello...”将消耗3秒钟。
稍微修改下上一节中Send.java的代码，让我们可以从命令行参数中输入任意字符作为消息。这个程序将给我们的工作队列安排消息，命名为NewTask.java：
String message = getMessage(argv);
channel.basicPublish("", "hello", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
一些封装方法来帮助我们从命令行参数中得到消息（简单来说就是将所有的命令行参数当做一条完整消息）：
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();
}
老的Recv.java程序也需要一些修改：他需要为消息中的每一个“.”伪造1秒钟的工作时间。称为Worker.java：
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");
}
}
};
boolean autoAck = true; // acknowledgment is covered below 消息确认，在后面会详细讲解
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
模拟任务执行：
private static void doWork(String task) throws InterruptedException {
for (char ch: task.toCharArray()) {
if (ch == '.') Thread.sleep(1000);
}
}
循环调度
使用Task Queue的优点之一就是可以轻松的进行并行工作。如果我们正在构建一个积压的工作，我们可以仅仅通过添加更多的workers来解决。
首先，同时运行两个worker实例，他们都会从队列中得到消息，但事实上是什么样的呢？让我们看一看：
在IDEA中运行两次Worker.java，然后他们两个都会处于等待消息状态。运行NewTask.java，并携带命令行参数，可以在Edit Configurations中设置Program arguements。下面为官方教程中的命令行版本：
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask First message.
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Second message..
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Third message...
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Fourth message....
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Fifth message.....
主要观察两个worker的输出：
worker1$ 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 'First message.'
[x] Received 'Third message...'
[x] Received 'Fifth message.....'
worker2$ 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将会按照顺序，以此发送每一条消息到每一个消费者。平均每个消费者是可以获得相同数量的消息的。这种分发消息的方式称为循环。
消息确认
完成一个任务需要消耗一定时间，你可能想知道如果一个消费者开始了一个很长的任务，在仅仅完成了一部分的时候，死掉了，将会发生什么。在我们当前的代码中，一旦RabbitMQ分发一条消息给消费者，立即就会将该条消息从内存中删除。这种情况下，如果你杀掉一个worker，我们将会丢失它正在操作的消息。我们也会失去所有分发给他的还未处理的消息。
但是我们不想丢失任何消息。如果worker死掉，我们期望这个任务被重新分发给另一个worker。
为了确保消息从来没有丢失，RabbitMQ支持消息确认（acknowledgments）。一个确认是从消费者处发送以告诉RabbitMQ指定的消息收到了，处理完成了，RabbitMQ可以删除它了。
如果一个消费者宕机（channel关闭，connection关闭，TCP连接丢失等），没有发送ack，RabbitMQ将会知道这条消息没有处理完成，将会重新排队。如果此时存在其它消费者，将会迅速转发给其它消费者。这样你就可以确保消息不会丢失，即使进程偶尔宕机。
这里不存在消息超时，RabbitMQ在消费者宕机后会重发消息。即使处理数据用了很长很长的时间这也是没有问题的。
默认的消息确认是被打开的。上面的例子中我们通过autoAck=true明确关闭了它。下面我们打开它，每当处理完一个任务，就发送回一个适当的确认消息。
channel.basicQos(1); // accept only one unack-ed message at a time (see below) 每次接收一个未处理消息
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);
}
}
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
使用现在的代码，我们可以保证即使在操作消息的时候通过CTRL+C关闭了一个消费者，也不会丢失消息。不久后，所有未处理完成的消息都会被重新发送。
Forgotten acknowledgment
忘记设置basicAck是很普通的事情，但是结果却很严重。当客户端退出（这可能听起来像随机分发）消息会被重新发送，但是RabbitMQ会吃掉越来越多的内容，因为它不会释放任何没有被确认的消息。
调试这种错误的使用rabbitmqctl来打印messages_unacknowledged的部分：
$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged
Listing queues ...
hello
...done.
消息持久化
我们学习了如何在消费者宕机的情况下保证数据不丢失。但是在RabbitMQ服务器宕机的情况下，数据依然是会丢失的。
当RabbitMQ退出或崩溃，它会忘记所有的队列和消息，除非你告诉它不要。两件事情来确保消息未丢失：我们需要标记队列和消息为持久化的。
首先，我们需要确保RabbitMQ从来不会丢失队列。因此我们需要声明队列为持久化的：
boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);
这行代码是没有问题的，但是在我们的环境下是错误的。这是因为我们已经定义了一个叫做hello的非持久化队列。RabbitMQ不允许重新定义已经存在的队列（使用不用参数）。这里有一个快速的方法 - 定义一个不同名字的队列，如task_queue：
boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
这个queueDeclare需要同时更改生产者和消费者的代码。
现在我们确保了task_queue在RabbitMQ重启的状态下也不会丢失。现在我们需要去标记我们的消息为持久化的 - 通过设置MessageProperties（实现了BasicProperties）的常量值：PERSISTENT_TEXT_PLAIN。
import com.rabbitmq.client.MessageProperties;
channel.basicPublish("", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
注意消息持久化：
标记了消息为持久化也不能完全保证消息不会丢失。尽管告诉了RabbitMQ将消息保存在磁盘中，RabbitMQ刚刚接收数据，还没有保存的时候，这个时间区间是无法持久化的。同事，RabbitMQ没有对每条消息都进行fsync(2) -- 也许仅仅保存在缓存中并没有真正写入硬盘。持久化并不健壮，但是对于处理简单的任务队列已经足够了。如果你需要更加强健的保证可以使用publisher confirms。
公平分发
你可能注意到有时候分发还是无法解决我们的某些问题。例如在某种情况下，有两个消费者，当所有的奇数消息非常大，偶数消息很小，一个消费者将会持续不断的工作，另一个消费者基本不工作。但是RabbitMQ并不知道这种情况，依然是依次分发。
这是因为RabbitMQ在消息进入队列是进行分发。并不探查消息的数量。仅仅是发送第n条消息给第n个消费者。
为了解决这个问题，我们可以使用basicQos方法，设置参数为prefetchCount = 1。这会告诉RabbitMQ每次只给一个消费者一条消息。或者说，不要在消费者正在处理和确认消息的时候发送新的消息给他们。相反，它将分发消息给下一个不忙的消费者。
int prefetchCount = 1;
channel.basicQos(prefetchCount);
注意队列的大小
如果所有的消费者都处于繁忙状态，队列会填满。可以添加更多的消费者或者其它方案。
Putting it all together
NewTask.java
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"; public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); String message = getMessage(argv); 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
import com.rabbitmq.client.*;
import java.io.IOException;
public class Worker {
private static final String TASK_QUEUE_NAME = "task_queue";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
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, 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);
}
}
};
channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
}
private static void doWork(String task) {
for (char ch : task.toCharArray()) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}
}
}