标签：
http://www.rabbitmq.com/tutorials/tutorial-two-java.html
在第一个教程中，我们通过一个命名队列来发送消息和接受消息。在这一节，我们将创建一个工作队列，在多个工作者之间，分发比较耗时的任务
工作队列主要是为了避免资源密集型任务的立即执行，然后一直等待它执行结束。相反，我们可以安排好任务，然后在执行。我们可以将一个任务封装成一个消息，发送到队列中。由工作者在后台取出任务然后执行。当有多个工作者时，他们共同处理这些任务。
在web应用中，当一次http请求需要处理复杂的任务时，工作队列将会变得非常有用
1：Preparation
在前面的教程中，我们发送一条消息“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秒。它会从消息队列中弹出一条消息，然后执行，所以我们叫他Work.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); } }
2：Round-robin dispatching(循环调度)
使用工作队列的一个优点就是可以并行的执行。如果我们需要处理积压的工作，可以通过这种方式添加更多的工作者，并且很容易扩展
首先，同时先运行两个Worker.java代码，它们都将从消息队列获取消息。我们需要开启3个控制端窗口。两个运行Worker.java代码，它们作为消费者
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
第三个窗口，我们会发布任务。一旦启动了consumer消费者之后，你就可以发布一些消息：
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会以此发送消息给下一个消费者。平均下来，每个消费者都会获得相同数量的消息。这种分发消息的方式就叫做Round-robin dispatching(循环调度)。
4：Message acknowledgment(消息确认)
一个任务可能花费几秒种的之间，当一个消费者开始执行一个耗时的任务，并且只执行了任务的一部分就死亡时，你或许希望知道究竟发生了什么。通过我们目前的代码，一旦RabbitMQ向向消费者传递消息，它会立即从内存中将其删除。在这种情况下，如果你杀死了一个worker，我们将会失去它正在处理的消息。我们同样会失去所有交给它，但还没来得及处理的消息。但是我们不希望丢失任何的tasks。如果一个worker死亡，我们希望该task被交给另一个worker来处理。
为了确保不丢失任何的message，RabbitMQ支持消息的确认。消费者会向RabbitMQ发送ack，来告诉RabbitMQ它已经接收，处理消息，RabbitMQ可以将其删除。
如果消费者没有发送ack就已经死亡，RabbitMQ会将其理解为消息没有被正确的处理，会将其从新发送给另一个消费者。通过这种方式，可以确保即便是worker偶然的死亡，也不会丢失任何消息。
不存在消息的超时。只有在worker的连接被关闭时，RabbitMQ才会将消息从新发送给下一个worker。即便是处理一个任务需要花费很长很长的时间，他也会正常的执行，而不会从新发送消息。
消息确认默认是开启的。在前面的例子中，我们通过autoAck=true，将其关闭。当我们执行一个task时，需要开启该功能，使worker可以发送一个正确的消息确认。
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正在处理一条消息时，强行将其关闭，也不会丢失任何的消息。在worker死亡后，所有未被确认的消息都将被从新发送。
5：Message durability(消息持久化)
我们已经学习了如何确保消费者死亡，任务也不会丢失。但是一旦我们的RabbitMQ服务停止，任务还是同样会丢失。
如果你不告诉RabbitMQ，当它退出或发生灾难关闭时，它会丢失队列里的消息。我们需要做两件事情来确保消息不会丢失：我们需要同时保证队列和消息时持久的。
首先：我们需要保证RabbitMQ不会丢失我们的队列。为了实现这个目的，我们需要将队列申明为永久的：
boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);
即便命令执行正常，在我们现有的程序中，它也不会起作用。那是因为我们已经存在一个非永久的队列“hello”。RabbitMQ不允许我们使用不同的参数来定义一个已经存在的队列，它会向我们返回一些错误信息。但是，通过使用不同的名字(eg：task_queue)，来定义一个新的队列，可以快速的解决这个问题。
boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
queueDeclare方法申明队列的变动，需要保证同时被应用在生产者和消费者两方
通过这种方式，我们可以确保task_queue队列在RabbitMQ服务重启后也不会丢失。现在我们需要把我们的消息也标记为永久的。通过设置MessageProperties(实现了BasicProperties)的值为PERSISTENT_TEXT_PLAIN可以实现这一目的。
channel.basicPublish("","task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
消息持久化注意事项
即便我们将消息标记为永久的，但这也不能完全保证消息就一定不会丢失。虽然RabbitMQ会将消息保存到硬盘上，但是在RabbitMQ接收到消息并且将其保存到硬盘上之间，仍然有一个短暂的时间片段。
6：Fair dispatch (公平调度)
可能你已经注意到了，分发工作并没有按照我们预期的方式来执行。例如：当有两个工作者时，所有的奇数任务都比较繁重，偶数任务都比较简单。这样就导致了一个工作者会一直处于忙碌状态，而另一个工作者几乎没有事情可做。RabbitMQ对于这些情况并不了解，它还会均匀的分配任务。
发生这样的原因是因为一旦有消息进入队列，RabbitMQ就会将消息进行分发。他不会去观察消费者发回的确认信息。它会盲目的将n条消息分发给n个消费者。为了杜绝这种情况，我们可以通过basicQos方法设置prefetchCount = 1。它会限制RabbitMQ在同一时间向一个工作者发送一条以上的消息。换句话说，它会等待一个工作者处理完或者接收到上一条消息的确认信息，才向它发送新的消息。否者，它会将其分发给下一个比较空闲的工作者。
int prefetchCount = 1;
channel.basicQos(prefetchCount);
注意：如果所有的工作者都处于忙碌的状态，你的队列可能会被填满，你需要时刻注意。或许你可以添加更多的工作者或者采用其他的策略。
7：Putting it all together(完整代码)
NewTask.java
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(); } //... }
Work.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); } } //... }
标签：
原文地址：http://www.cnblogs.com/d392110851/p/5357056.html
