rabbitmq學習記錄（五）交換機Exchange-fanout

以前學習的都是一條消息發給一個消費者，下面開始記錄如何把一條信息發給多個消費者

這邊咱們用到了交換機Exchange

交換機模式：fanout

模式特色：生產者把消息發送給Exchange以後，Exchange則會把這些消息添加到與本身綁定的全部隊列之中，監聽這些隊列的消費者就能夠收到這些消息。

注：Exchange並不能保存信息，若是沒有綁定的隊列，那麼生產者發送數據就會丟失，只有隊列才能存儲生產者的消息。

生產者：聲明交換機後指定交換機模式fanout

package com.example.demo.queue.exchangeToQueue.fanout;

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 {

	// exchange名稱
	private static final String EXCHANGE_NAME = "exchange_fanout"; 
	
	public static void main(String[] args) {
		Connection connection = null;
		Channel channel = null;
		try {
			// 獲取鏈接
			connection = ConnectionUtil.getConnection();
			// 建立通道
			channel = connection.createChannel();
			// 聲明交換機
			channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);
			// 生產者發送的信息
			String msg = "msg from producer:";
			for(int i=0;i<10;i++) {
				msg = "msg from producer :" + i;
				System.out.println("send msg : "+msg);
				// 發送信息
				channel.basicPublish(EXCHANGE_NAME, "", null, msg.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：聲明隊列fanout_exchange_to_queue_01並綁定到交換機exchange_fanout

package com.example.demo.queue.exchangeToQueue.fanout;

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 Consumer01 {

	// 交換機名稱
	private static final String EXCHANGE_NAME = "exchange_fanout"; 
	
	// 監聽隊列名稱
	private static final String QUEUE_NAME = "fanout_exchange_to_queue_01";
	
	public static void main(String[] args) {
		try {
			// 獲取鏈接
			Connection connection = ConnectionUtil.getConnection();
			// 建立通道
			final Channel channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			// 綁定隊列到交換機
			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
			// 定義消費者
			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);
					System.out.println("[1]:deal msg successful.");
				}
				
			};
			// 接收信息
			channel.basicConsume(QUEUE_NAME, true, consumer);
			
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
}

消費者2：聲明隊列fanout_exchange_to_queue_02並綁定到交換機exchange_fanout

package com.example.demo.queue.exchangeToQueue.fanout;

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 EXCHANGE_NAME = "exchange_fanout"; 
	
	// 監聽隊列名稱
	private static final String QUEUE_NAME = "fanout_exchange_to_queue_02";
	
	public static void main(String[] args) {
		try {
			// 獲取鏈接
			Connection connection = ConnectionUtil.getConnection();
			// 建立通道
			final Channel channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			// 綁定隊列到交換機
			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
			// 定義消費者
			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);
					System.out.println("[2]:deal msg successful.");
				}
			};
			// 接收信息
			channel.basicConsume(QUEUE_NAME, true, consumer);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
}

消費者3：聲明隊列fanout_exchange_to_queue_03，沒有綁定到交換機exchange_fanout

package com.example.demo.queue.exchangeToQueue.fanout;

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 Consumer03 {

	// 交換機名稱
	private static final String EXCHANGE_NAME = "exchange_fanout"; 
	
	// 監聽隊列名稱
	private static final String QUEUE_NAME = "fanout_exchange_to_queue_03";
	
	public static void main(String[] args) {
		try {
			// 獲取鏈接
			Connection connection = ConnectionUtil.getConnection();
			// 建立通道
			final Channel channel = connection.createChannel();
			// 聲明隊列
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			// 綁定隊列到交換機
//			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
			// 定義消費者
			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("[3]:receive msg:"+msg);
					System.out.println("[3]:deal msg successful.");
				}
			};
			// 接收信息
			channel.basicConsume(QUEUE_NAME, true, consumer);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
}

此次，涉及到了交換機，而在消費者類中中並無聲明交換機，因此須要先執行生產者類的main方法

能夠看到生產者已經成功發送消息給交換機，交換機也成功建立了，可是消息數據丟失了，緣由就是上面說的，交換機沒法保存從生產者那接收到的信息，只有隊列能夠保存消息。

接下來，咱們就能夠依次執行三個消費者類的main方法，以後再執行生產者類的main方法。

消費者1終端：

消費者2終端：

消費者3終端：