RabbitMQ使用场景练习：发布/订阅(三)

• 发布/订阅

     即实现单点发送消息，多点接收。使用fanout转发器，广播到所有它知道的队列上

• 注意要点

fanout转发器中不需要routingKey，指定也无效

创建fanout转发器：

channel.exchangeDeclare("fanout_logs", "fanout");
channel.basicPublish("fanout_logs", "" , null, SerializationUtils.serialize(object));

临时队列的创建：

//创建一个临时队列，返回队列名称
String queueName=channel.queueDeclare().getQueue();

转发器与队列的绑定：

 //绑定临时队列和转发器logs
channel.queueBind(queueName, "fanout_logs", "");

• 消息发送类

package com.demo.mq.rabbitmq.example03;

import java.io.IOException;
import java.io.Serializable;

import org.apache.commons.lang3.SerializationUtils;

import com.demo.mq.rabbitmq.MqManager;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 发送消息类
 * @author sheungxin
 *
 */
public class Send{

	/**
	 * 发送消息，fanout转发器，广播到所有在此转发器上的队列
	 * @param object 消息主体
	 * @throws IOException
	 */
	public static void sendAToB(Serializable object) throws Exception{
		Connection conn=MqManager.newConnection();
		Channel channel=conn.createChannel();
		channel.exchangeDeclare("fanout_logs", "fanout");
		channel.basicPublish("fanout_logs", "" , null, SerializationUtils.serialize(object));
		System.out.println("A Send :'"+object+"'");
		channel.close();
		conn.close();
	}
	
	public static void main(String[] args) throws Exception {
		for(int i=0;i<4;i++){
			sendAToB("Hello World "+i+"!");
		}
	}
}

• 消息接收类

package com.demo.mq.rabbitmq.example03;

import java.io.IOException;
import org.apache.commons.lang3.SerializationUtils;
import com.demo.mq.rabbitmq.MqManager;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 接收消息类
 * @author sheungxin
 *
 */
public class Recv {
	
	/**
	 * 用于接收消息，创建一个临时队列，绑定在转发器fanout上
	 * @throws Exception
	 */
	public static void recvAToB() throws Exception{
		Connection conn=MqManager.newConnection();
		Channel channel=conn.createChannel();
		channel.exchangeDeclare("fanout_logs", "fanout");
		//创建一个临时队列
		String queueName=channel.queueDeclare().getQueue();
		//绑定临时队列和转发器logs
		channel.queueBind(queueName, "fanout_logs", "");
		Consumer consumer=new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException{
				String mes=SerializationUtils.deserialize(body);
				System.out.println("B Received :'"+mes+"'...");
			}
		};
		//关闭自动应答机制，默认开启；这时候需要手动进行应该
		channel.basicConsume(queueName, true, consumer);
	}
	
	public static void main(String[] args) throws Exception {
		recvAToB();
	}

}