上一篇博文中簡單介紹了一下RabbitMQ的基礎知識，並寫了一個經典語言入門程式——HelloWorld。本篇博文中我們將會建立一個工作佇列用來在工作者（consumer）間分發耗時任務。同樣是翻譯的官網例項。

工作佇列

       在前一篇博文中，我們完成了一個簡單的對宣告的佇列進行傳送和接受訊息程式。下面我們將建立一個工作佇列，來向多個工作者（consumer）分發耗時任務。

       工作佇列（又名：任務佇列）的主要任務是為了避免立即做一個資源密集型的卻又必須等待完成的任務。相反的，我們進行任務排程：將任務封裝為訊息併發給佇列。在後臺執行的工作者（consumer）將其取出，然後最終執行。當你執行多個工作者（consumer），佇列中的任務被工作進行共享執行。

       這樣的概念對於在一個HTTP短連結的請求視窗中處理複雜任務的web應用程式，是非常有用的。

準備

       傳送端（生產者）：NewTask.java

public class NewTask {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] args) throws IOException {
		/**
		 * 建立連線連線到MabbitMQ
		 */
		ConnectionFactory factory = new ConnectionFactory();
		// 設定MabbitMQ所在主機ip或者主機名
		factory.setHost("127.0.0.1");
		// 建立一個連線
		Connection connection = factory.newConnection();
		// 建立一個頻道
		Channel channel = connection.createChannel();
		// 指定一個佇列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		// 傳送的訊息
		String message = "Hello World...";
		// 往佇列中發出一條訊息
		channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
		System.out.println(" [x] Sent '" + message + "'");
		// 關閉頻道和連線
		channel.close();
		connection.close();
	}
}
 

       工作者（消費者）Worker.java

public class Worker {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] argv) throws IOException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		// 開啟連線和建立頻道，與傳送端一樣
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		// 宣告佇列，主要為了防止訊息接收者先執行此程式，佇列還不存在時建立佇列。
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
		
		// 建立佇列消費者
		final Consumer consumer = new DefaultConsumer(channel) {
			  @Override
			  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
			    String message = new String(body, "UTF-8");

			    System.out.println(" [x] Received '" + message + "'");
			    System.out.println(" [x] Proccessing... at " +new Date().toLocaleString());
			    try {
			    	for (char ch: message.toCharArray()) {
				        if (ch == '.') {
				        	Thread.sleep(1000);
				        }
				    }
				} catch (InterruptedException e) {
				} finally {
			      System.out.println(" [x] Done! at " +new Date().toLocaleString());
			    }
			  }
			};
			channel.basicConsume(QUEUE_NAME, true, consumer);
	}
}
 

任務分發機制

       正主來了。。。下面開始介紹各種任務分發機制。

Round-robin（輪詢分發）

修改一下NewTask，使用for迴圈模擬多次傳送訊息的過程：

		for (int i = 0; i < 5; i++) {
			// 傳送的訊息
			String message = "Hello World"+Strings.repeat(".", i);
			// 往佇列中發出一條訊息
			channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
			System.out.println(" [x] Sent '" + message + "'");
		}

       我們先啟動1個生產者例項，2個工作者例項，看一下如何執行：

       從上述的結果中，我們可以得知，在預設情況下，RabbitMQ將逐個傳送訊息到在序列中的下一個消費者(而不考慮每個任務的時長等等，且是提前一次性分配，並非一個一個分配)。平均每個消費者獲得相同數量的訊息。這種方式分發訊息機制稱為Round-Robin（輪詢）。

Fair dispatch（公平分發）

       您可能已經注意到，任務分發仍然沒有完全按照我們想要的那樣。比如：現在有2個消費者，所有的奇數的訊息都是繁忙的，而偶數則是輕鬆的。按照輪詢的方式，奇數的任務交給了第一個消費者，所以一直在忙個不停。偶數的任務交給另一個消費者，則立即完成任務，然後閒得不行。而RabbitMQ則是不瞭解這些的。

       這是因為當訊息進入佇列，RabbitMQ就會分派訊息。它不看消費者為應答的數目，只是盲目的將第n條訊息發給第n個消費者。

       為了解決這個問題，我們使用basicQos( prefetchCount = 1)方法，來限制RabbitMQ只發不超過1條的訊息給同一個消費者。當訊息處理完畢後，有了反饋，才會進行第二次傳送。

int prefetchCount = 1;
channel.basicQos(prefetchCount);

       整體程式碼如下：生產者NewTask.java

public class NewTask {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] args) throws IOException {
		/**
		 * 建立連線連線到MabbitMQ
		 */
		ConnectionFactory factory = new ConnectionFactory();
		// 設定MabbitMQ所在主機ip或者主機名
		factory.setHost("127.0.0.1");
		// 建立一個連線
		Connection connection = factory.newConnection();
		// 建立一個頻道
		Channel channel = connection.createChannel();
		// 指定一個佇列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		int prefetchCount = 1;
		//限制發給同一個消費者不得超過1條訊息
		channel.basicQos(prefetchCount);
		for (int i = 0; i < 5; i++) {
			// 傳送的訊息
			String message = "Hello World"+Strings.repeat(".",5-i)+(5-i);
			// 往佇列中發出一條訊息
			channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
			System.out.println(" [x] Sent '" + message + "'");
		}
		// 關閉頻道和連線
		channel.close();
		connection.close();
	}
}

       消費者Worker.java

public class Worker {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] argv) throws IOException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		// 開啟連線和建立頻道，與傳送端一樣
		Connection connection = factory.newConnection();
		final Channel channel = connection.createChannel();

		// 宣告佇列，主要為了防止訊息接收者先執行此程式，佇列還不存在時建立佇列。
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
		channel.basicQos(1);//保證一次只分發一個
		// 建立佇列消費者
		final Consumer consumer = new DefaultConsumer(channel) {
			  @Override
			  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
			    String message = new String(body, "UTF-8");

			    System.out.println(" [x] Received '" + message + "'");
			    try {
			    	for (char ch: message.toCharArray()) {
				        if (ch == '.') {
				        	Thread.sleep(1000);
				        }
				    }
				} catch (InterruptedException e) {
				} finally {
			      System.out.println(" [x] Done! at " +new Date().toLocaleString());
			      channel.basicAck(envelope.getDeliveryTag(), false);  
			    }
			  }
			};
			channel.basicConsume(QUEUE_NAME, false, consumer);
	}
}