订阅模式 - fanout

package com.shi.ps;

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

import org.junit.Test;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
import com.shi.util.RabbitMqUtils;

/**
 * 订阅模式
 * @author SHF
 * @version 创建时间：2018年7月4日  下午2:43:50
 */
public class PublishSubscribe {
	
	//交换机名称
	private final static String EXCHANGE_NAME = "exchange_fanout";
	
	//队列名称
	private final static String QUEUE_1 ="exchange_queue_1";
	private final static String QUEUE_2 ="exchange_queue_2";
	
	/**
	 * 生产者  - 订阅模式
	 * @author SHF
	 * @version 创建时间：2018年7月4日  下午2:59:42
	 * @throws TimeoutException 
	 * @throws IOException 
	 */
	@Test
	public void send() throws IOException, TimeoutException {
		//1  获取链接以及mq通道
		  Connection connection = RabbitMqUtils.getConnection();
		  Channel channel = connection.createChannel();
		  
		//2  申明exchange （交换机）  "fanout":交换机的类型
		 channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
		  
		//3 消息内容
		 String message = "施爷 exchange_fanout  来消息了....";
		 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
		 
		//4 关闭通道及连接
		 channel.close();
		 connection.close();
	}
	
	/**
	 * 消费者1 - 订阅模式
	 * @author SHF
	 * @version 创建时间：2018年7月4日  下午3:12:31
	 * @throws TimeoutException 
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws ConsumerCancelledException 
	 * @throws ShutdownSignalException 
	 */
	@Test
	public void reic1() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		//1  获取链接 及mq通道
		Connection connection = RabbitMqUtils.getConnection();
		Channel channel = connection.createChannel();
		
		//2  申明队列
		channel.queueDeclare(QUEUE_1, false, false, false, null);
		
		//3  绑定队列到交换机
		channel.queueBind(QUEUE_1, EXCHANGE_NAME, "");
		
		//4  同一时刻服务器只会发送一条消息给消费者
		channel.basicQos(1);
		
		//5  定义队列的消费者
		QueueingConsumer consumer = new QueueingConsumer(channel);
		
		//6  监听队列，手动返回完成(false)
		channel.basicConsume(QUEUE_1, false,consumer);
		
		//7  获取消息
		while(true) {
			Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());
			System.out.println(" x :exchange_queue_1 接收到消息：" + message);
			Thread.sleep(10);
			
			//手动返回  告诉队列消息已经成功接收
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
		}
	}
	
	/**
	 * 消费者2 - 订阅模式
	 * @author SHF
	 * @version 创建时间：2018年7月4日  下午3:37:49
	 * @throws TimeoutException 
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws ConsumerCancelledException 
	 * @throws ShutdownSignalException 
	 */
	@Test
	public void reiv2 () throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		//1   获取链接 借mq通道
		Connection connection = RabbitMqUtils.getConnection();
		Channel channel = connection.createChannel();
		
		//2 申明队列
		channel.queueDeclare(QUEUE_2, false, false, false, null);
		
		//3 绑定队列到交换机
		channel.queueBind(QUEUE_2, EXCHANGE_NAME, "");
		
		//4 同一时刻服务器只发送一条消息给消费者
		channel.basicQos(1);
		
		//5 定义队列的消费者
		QueueingConsumer consumer = new QueueingConsumer(channel);
		
		//6 监听队列，手动返回完成(false:表示需要手动完成)
		channel.basicConsume(QUEUE_2, false, consumer);
		
		//7 获取消息
		while(true) {
			Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());
			System.out.println(" x :exchange_queue_2  接收到消息：" + message);
			Thread.sleep(10);
			
			//手动返回  告诉队列消息已经成功接收
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
		}
	}
}