柯南君：看大数据时代下的IT架构（5）消息队列之RabbitMQ--案例（Work Queues起航）

二、Work Queues（using the Java Client) 走起

在第上一个教程中我们写程序从一个命名队列发送和接收消息。在这一次我们将创建一个工作队列,将用于分发耗时的任务在多个工作者（worker）之间。

背后的主要思想工作队列(又名:任务队列)是为了避免立即做一个资源密集型任务,不得不等待它完成。相反,我们安排的任务要做。我们封装任务作为消息并将其发送到一个队列。工作进程在后台运行将流行的任务和最终执行的工作。当您运行许多worker的任务将在他们之间共享。这个概念是特别有用的web应用程序中处理复杂的任务是不可能在一个短的HTTP请求窗口。
为了避免等待一些占用大量资源、时间的操作。当我们把任务（Task）当作消息发送到队列中，一个运行在后台的工作者（worker）进程就会取出任务然后处理。当你运行多个工作者（workers），任务就会在它们之间共享。

三、Preparation（预备）

在RabbitMQ系列教程中，在前一章柯南君：看大数据时代下的IT架构（4）消息队列之RabbitMQ--案例（Helloword起航）

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);
}
return words.toString();
}

我们的老的Recv.java程序也需要一些变化:

while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
doWork(message);
System.out.println(" [x] Done");
}

private static void doWork(String task) throws InterruptedException {
for (char ch: task.toCharArray()) {
if (ch == '.') Thread.sleep(1000);
}
}

编译它们 (工作目录中的jar文件):

$ javac -cp rabbitmq-client.jar NewTask.java Worker.java

使用一个任务队列的优点之一是能够轻易平行的工作。如果我们建立一个积压的工作,我们可以添加更多的任务（worker）,这样,很容易。
首先,让我们试着两个任务（worker）实例同时运行。他们都将从队列中获取消息,但究竟如何?让我们来看看。你需要三个主机开放。
两人将运行工计划。这些游戏机将我们两个消费者- C1和C2。

shell1$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
Worker
[*] Waiting for messages. To exit press CTRL+C

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

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.....

shell1$ 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.....'

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向客户传递一个消息立即从内存中删除。在这种情况下,如果你kill（杀）了一个任务（worker）,
我们将失去刚刚处理的消息。我们也会失去所有的消息被派往这个特殊的工人,但尚未处理。　
但是我们不想失去任何任务。如果一个工作者（worker）死亡,我们想要交付的任务到另一个worker。
为了确保消息是从来都不会迷失的,RabbitMQ支持消息应答。发送ack(knowledgement)从消费者告诉RabbitMQ特定的消息已经收到,
处理和RabbitMQ是免费的,删除它。如果一个消费者停止没有发送ack,RabbitMQ会明白一个消息

QueueingConsumer consumer = new QueueingConsumer(channel);
boolean autoAck = false;
channel.basicConsume("hello", autoAck, consumer);

while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
//...
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}

使用这段代码,我们可以确信,即使你杀了一个工作者（worker）使用CTRL + C处理消息时,没有将丢失。工作者（worker）死亡后不久所有未得到确认的消息将被发送。

问题：
   消息响应

六、Message durability

我们已经学会了如何确保即使消费者死亡,任务也不会丢失。但是如果RabbitMQ服务器停止我们的任务仍将失去。

RabbitMQ退出或崩溃时它会忘记队列和消息,除非你告诉它不要。两件事必须确保消息不会丢失:我们需要两个队列和消息标记为耐用。
首先,我们需要确保RabbitMQ永远不会失去我们的队列。为了这样做,我们需要声明它经久耐用:
channel.queueDeclare("hello", durable, false, false, null);
尽管这个命令本身是正确的,它不会在我们目前的设置工作。这是因为我们已经定义了一个名为hello的队列不耐用。RabbitMQ不允许您重新定义现有队列具有不同参数并返回一个错误的任何程序,试图这样做。但有一个快速解决方案——让我们声明一个队列具有不同名称,例如task_queue:
channel.queueDeclare("task_queue", durable, false, false, null);
这queueDeclare改变需要应用于生产者和消费者的代码。在这一点上我们确信task_queue队列不会丢失,即使RabbitMQ重启。现在我们需要我们的消息标记为持久性——通过设置MessageProperties PERSISTENT_TEXT_PLAIN(实现BasicProperties)的价值。
channel.basicPublish("", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
问题：
消息标记为持久性并不能完全保证信息不会丢失。虽然它告诉RabbitMQ将消息保存到磁盘,还有一个短的时间窗口当RabbitMQ已经接受消息和尚未保存它。
RabbitMQ也不做fsync(2)为每个消息——它可能只是保存到缓存和不写入磁盘。持久性保证不强,但它是足够为我们简单的任务队列。
如果你需要一个更强的保证,那么你可以使用发布者证实。

boolean durable = true;

boolean durable = true;

import com.rabbitmq.client.MessageProperties;

七、Fair dispatch

您可能已经注意到,调度仍然不会完全按照我们想要的工作。例如在两名工人（worker）的情况,当所有奇怪的消息是沉重的,甚至消息是光,一名工人将不停地忙,另一个几乎不做任何工作。RabbitMQ并不了解,仍将均匀调度信息。这仅仅是因为RabbitMQ分派消息进入队列的消息的时候。它不看看消费者未得到确认的消息的数量。只是盲目地分派每n个消息到n个消费者。

为了打败,我们可以使用basicQos prefetchCount = 1设置方法。这告诉RabbitMQ不给多个消息到一个工人。或者,换句话说,不要派遣工人的新消息,直到处理和承认了前一个。相反,它会分派到下一个工人,不是仍然很忙。
channel.basicQos(prefetchCount);

int prefetchCount = 1;

问题：
如果所有的工人们正忙着,你的队列可以填满。你要留意,也许添加更多的工人,或有其他一些策略。

八、Putting
it all together

Final code of our NewTask.java class:

import java.io.IOException;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
public class NewTask {

private static final String TASK_QUEUE_NAME = "task_queue";

public static void main(String[] argv)
throws java.io.IOException {

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());
System.out.println(" [x] Sent '" + message + "'");

channel.close();
connection.close();
}
//...
}

import java.io.IOException;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;

public class Worker {

private static final String TASK_QUEUE_NAME = "task_queue";

public static void main(String[] argv)
throws java.io.IOException,
java.lang.InterruptedException {

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
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);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(TASK_QUEUE_NAME, false, consumer);

while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());

System.out.println(" [x] Received '" + message + "'");
doWork(message);
System.out.println(" [x] Done" );

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
//...
}