• 我们稍微改一下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(); }

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

在第三个我们将发布新的任务。一旦你开始了消费者可以发布几条消息:

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将每个消息发送到下一个消费者,在序列。平均每个消费者将获得相同数量的信息。这种方式称为循环分配消息。试试这三个或更多的worker。

五、Message acknowledgment

 做一个任务可能需要几秒钟。如果一个消费者开始漫长的任务而死,只有部分完成，你可能想知道到底发生什么事情了？
在我们当前的代码情况下,一旦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); }

六、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(); } //... } 

(NewTask.java source)

And our Worker.java:

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); } } //... }