rabbitmq学习记录（五）交换机Exchange-fanout

时间 2019-11-05

标签 rabbitmq 学习记录交换机 exchange fanout 栏目 RabbitMQ 繁體版

原文原文链接

以前学习的都是一条消息发给一个消费者，下面开始记录如何把一条信息发给多个消费者java

这边咱们用到了交换机Exchangeide

交换机模式：fanout学习

模式特色：生产者把消息发送给Exchange以后，Exchange则会把这些消息添加到与本身绑定的全部队列之中，监听这些队列的消费者就能够收到这些消息。code

注：Exchange并不能保存信息，若是没有绑定的队列，那么生产者发送数据就会丢失，只有队列才能存储生产者的消息。blog

生产者：声明交换机后指定交换机模式fanoutrabbitmq

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_fanoutget

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_fanoutit

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方法io

能够看到生产者已经成功发送消息给交换机，交换机也成功建立了，可是消息数据丢失了，缘由就是上面说的，交换机没法保存从生产者那接收到的信息，只有队列能够保存消息。

接下来，咱们就能够依次执行三个消费者类的main方法，以后再执行生产者类的main方法。

消费者1终端：

消费者2终端：

消费者3终端：