rabbitmq學習記錄（三）工做隊列-輪詢分配

工做隊列：一個生產者對應多個消費者，生產者直接將消息發送到rabbitmq的隊列之中。

消息分配模式：輪詢分配（默認）

        即無論消費者處理信息的效率，隊列給全部消費者輪流發送一條信息，直至消息發送完畢

假如：生產者發送了10條信息

那麼

消費者A收到的信息：1，4，7，10

消費者B收到的信息：2，5，8

消費者C收到的信息：3，6，9

生產者類：

package com.example.demo.queue.work;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.example.demo.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Producer {

	// 隊列名稱
	private static final String QUEUE_NAME="work_queue";
	
	public static void main(String[] args) {
		Connection connection = null;
		Channel channel = null;
		try {
			// 獲取鏈接
			connection = ConnectionUtil.getConnection();
			// 建立通道
			channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, false, false, false, null);
			// 生產者發送的信息
			String sendMsg = "msg from producer";
			for(int i=0 ;i<10; i++) {
				sendMsg = "msg from producer :" + i;
				// 發送信息
				channel.basicPublish("", QUEUE_NAME, null, sendMsg.getBytes());
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 關閉通道
			try {
				channel.close();
			} catch (IOException e) {
				e.printStackTrace();
			} catch (TimeoutException e) {
				e.printStackTrace();
			}
			// 關閉鏈接
			try {
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
	}
	
}

消費者1：

package com.example.demo.queue.work;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.example.demo.utils.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class Consumer01 {

	// 隊列名稱
	private static final String QUEUE_NAME="work_queue";
	
	public static void main(String[] args) {
		Connection connection = null;
		Channel channel = null;
		try {
			// 獲取鏈接
			connection = ConnectionUtil.getConnection();
			// 建立通道
			channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, false, false, false, null);
			// 定義消費者
			DefaultConsumer consumer = new DefaultConsumer(channel) {

				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
					String msg = new String(body,"UTF-8");
					System.out.println("[1]:receive msg:"+msg);
					try {
						Thread.sleep(1000);
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						System.out.println("[1]:deal msg successful.");
					}
				}
			};
			// 接收信息
			channel.basicConsume(QUEUE_NAME, true, consumer);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 關閉通道
//			try {
//				channel.close();
//			} catch (IOException e) {
//				e.printStackTrace();
//			} catch (TimeoutException e) {
//				e.printStackTrace();
//			}
			// 關閉鏈接
//			try {
//				connection.close();
//			} catch (IOException e) {
//				e.printStackTrace();
//			}
		}
	}
	
}

消費者2：

package com.example.demo.queue.work;

import java.io.IOException;

import com.example.demo.utils.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class Consumer02 {

	// 隊列名稱
	private static final String QUEUE_NAME="work_queue";
	
	public static void main(String[] args) {
		Connection connection = null;
		Channel channel = null;
		try {
			// 獲取鏈接
			connection = ConnectionUtil.getConnection();
			// 建立通道
			channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, false, false, false, null);
			// 定義消費者
			DefaultConsumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
					String msg = new String(body,"UTF-8");
					System.out.println("[2]:receive msg:"+msg);
					try {
						Thread.sleep(2000);
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						System.out.println("[2]:deal msg successful.");
					}
				}
			};
			// 接收信息
			channel.basicConsume(QUEUE_NAME, true, consumer);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 關閉通道
//			try {
//				channel.close();
//			} catch (IOException e) {
//				e.printStackTrace();
//			} catch (TimeoutException e) {
//				e.printStackTrace();
//			}
			// 關閉鏈接
//			try {
//				connection.close();
//			} catch (IOException e) {
//				e.printStackTrace();
//			}
		}
	}
	
}

執行順序：

        先執行兩個消費者（Consumer01，Consumer02）類的main方法，再執行生產者（Producer）的main方法便可

順序反過來亦可，可是可能會影響到輪詢分配的效果

消費者1終端：

消費者2終端：