rabbitmq学习2:Work Queues
在前面的已经提到了一对一的情况;现在一个生产者与多个消费者的情况（Work Queues）。
Work Queues的示意图如下：
?
对于上图的模型中对于c端的worker来说。RabbitMQ服务器可能一直发送多个消息给一个worker，而另一个可能几乎不做任何事情。这样就会导致一个worker很忙，而另一个却很空闲。这种情况可能都不想出现。如何解决这个问题呢。当然最理想的情况是均匀分配消息给每个worker。我们可能通过
?
P端的程序如下：
package com.abin.rabbitmq;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());//持久化消息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();}}??? 多次运行此程序并传入的参数分别为“package com.abin.rabbitmq;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.QueueingConsumer;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");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);//告诉RabbitMQ同一时间给一个消息给消费者/* We're about to tell the server to deliver us the messages from the queue. * Since it will push us messages asynchronously, * we provide a callback in the form of an object that will buffer the messages * until we're ready to use them. That is what QueueingConsumer does.*/QueueingConsumer consumer = new QueueingConsumer(channel);/* 把名字为TASK_QUEUE_NAME的Channel的值回调给QueueingConsumer,即使一个worker在处理消息的过程中停止了，这个消息也不会失效*/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);//下一个消息}}private static void doWork(String task) throws InterruptedException {for (char ch : task.toCharArray()) {if (ch == '.')Thread.sleep(1000);//这里是假装我们很忙}}}
? 开启两个worker分别运行。运行结果如：
c1的结果：
[*] Waiting for messages. To exit press CTRL+C [x] Received 'First message' [x] Received 'Third message' [x] Received 'Fifth message'
?c2的结果
[*] Waiting for messages. To exit press CTRL+C [x] Received 'Second message' [x] Received 'Fourth message'?
?
?