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'