rabbitmq学习2:Work Queues

在前面的已经提到了一对一的情况;现在一个生产者与多个消费者的情况（Work Queues）。
Work Queues的示意图如下：

 

对于上图的模型中对于c端的worker来说。RabbitMQ服务器可能一直发送多个消息给一个worker，而另一个可能几乎不做任何事情。这样就会导致一个worker很忙，而另一个却很空闲。这种情况可能都不想出现。如何解决这个问题呢。当然最理想的情况是均匀分配消息给每个worker。我们可能通过channel . basicQos(1)方法（prefetchCount = 1 ）来设置同一时间每次发给一个消息给一个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();
    }
}

    多次运行此程序并传入的参数分别为“First message ”，“Secondmessage ”，“Third message ”，“Fourth message ”，“Fifth message ”

 

C端的程序如下：

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'

 

 

评论

4 楼 ZZX19880809 2014-11-01  

根本就没有. 应该输入first message.

3 楼 jiaofuyou 2014-03-19  

独孤日日也 写道

我亲自试验了并没有实现啊，第一个worker输出：

Waiting for messages. To exit press CTRL+C

[x] Received '1 message'
[x] Done
[x] Received '2 message'
[x] Done
[x] Received '3 message'
[x] Done
第二个worker输出：

Waiting for messages. To exit press CTRL+C

这是为什么呢

我是用PHP测试的，是可以的

2 楼 独孤日日也 2014-03-18  

我亲自试验了并没有实现啊，第一个worker输出：

Waiting for messages. To exit press CTRL+C

[x] Received '1 message'
[x] Done
[x] Received '2 message'
[x] Done
[x] Received '3 message'
[x] Done
第二个worker输出：

Waiting for messages. To exit press CTRL+C

这是为什么呢

1 楼 jiaofuyou 2014-03-13  

想问个问题，象这种任务分发的工作队列，你举的例子是一个队列被多个消费者处理，那有没有办法，多个队列的消息由一个消费者来处理呢？