码迷,mamicode.com
首页 > 编程语言 > 详细

RabbitMQ - Publish/Subscribe in Java

时间:2014-07-12 16:52:58      阅读:410      评论:0      收藏:0      [点我收藏+]

标签:rabbitmq   java   publish   

这次我们试试publish / subscribe模式,

也就是将一个消息发送给多个consumer。



这里用一个简单的小程序来说明publish / subscribe。

由一个provider提供消息,这个消息会被多个consumer接收。

consumer对同一个消息做出不同的反应,比如打印、保存到文件、数据库什么的。



之前的例子可能会给人这种感觉:

producer将消息发送到队列中,消息缓冲在队列中,consumer从队列获得消息。


但这并不正确。

在rabbit中,producer从来不会直接将消息发送到队列中。

producer根本无从得知消息是否会发送到某个队列中。



事实上,producer只能将消息发送到exchange中。

这么一说虽然感觉多了个东西,但exchange并不复杂。

exchange只是从producer获取消息并将消息推送到队列中。

bubuko.com,布布扣


但为什么多了这么个步骤?

比如exchange收到消息后,它应该将消息推送给某个特定的队列? 或者可以将消息推送给多个队列? 再或者直接抛弃该消息? 

这些规则取决于exchange的类型。



以下是一些可用的exchange type(org.springframework.amqp.core.ExchangeTypes):

	public static final String DIRECT = "direct";

	public static final String TOPIC = "topic";

	public static final String FANOUT = "fanout";

	public static final String HEADERS = "headers";

	public static final String SYSTEM = "system";


我们可以用以下方式定义一个exchange:

channel.exchangeDeclare("logs", "fanout");

正如其名,fanout就是将收到的消息发送给所有可访问的队列。



如何查看已定义的exchange?

查看已定义的exchange,我们可以用rabbitmqctl list_exchanges命令,如图:

bubuko.com,布布扣

图中名为amq.*和没有名字的exchange都是默认自带的。

(PS:之前的例子中我们还没有用到exchange的概念,但仍然成功地将消息发送到了队列中。

这是因为我们使用的是默认的exchange。)



我们需要将消息发送到指定的exchange中。

basicPublish的第一个参数就是exchange的名称(重写的几个都是)。

空字符串表示默认的exchange:

channel.basicPublish( "logs", "", null, message.getBytes());



队列的命名很重要,比如多个worker共享一个队列,producer和consumer的关系用队列名维系。

但并不是所有的场景都需要我们亲自去命名。

比如我们需要获得所有消息,而不是它的某个子集。

或者我们更关心最新的消息,而不是更早放到队列的那些。


我们需要让server随机命名队列,并且队列在consumer连接断开时自动删除。


我们只需要一行代码来做这些:

String queueName = channel.queueDeclare().getQueue();

调用不带参数的queueDeclare()可以创建一个临时队列。



到此我们就创建好了exchange和队列。

我们需要用什么东西将他们联系起来,这个东西叫作"binding"。


通过以下代码将他们联系起来:

channel.queueBind(queueName, "logs", "");


正如查看exchange那样,我们可以用rabbitmqctl list_bindings命令查看binding。

如图:

bubuko.com,布布扣


从producer到queue的关系图如下:

bubuko.com,布布扣


写了个Channel静态工厂,写的不好。

我打算在静态初始化块中定义两个exchange:

final class ChannelFactory {
	
	//consumer的temporary queue与这两个exchange绑定
	final static String EXCHANGE_NAME = "log";
	final static String EXCHANGE_NAME_ = "log2";
	
	private static final ConnectionFactory factory = new ConnectionFactory();
	
	static{
		try {
			Channel temp = getChannel();
			temp.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.FANOUT);
			temp.exchangeDeclare(EXCHANGE_NAME_, ExchangeTypes.FANOUT);
			closeChannel(temp);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private ChannelFactory() {
	}

	public static Channel getChannel() {
		try {
			return factory.newConnection().createChannel();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	public static Channel getChannel(int channelNumber) {
		try {
			return factory.newConnection().createChannel();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	public static void closeChannel(Channel channel) {
		try {
			channel.close();
			channel.getConnection().close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
}


producer类,同一个producer给两个exchange发消息:

class Publisher {
	public static void main(String[] args) throws IOException {

		Channel channel = ChannelFactory.getChannel();

		String message = "Here is the content";
		channel.basicPublish(ChannelFactory.EXCHANGE_NAME, StringUtils.EMPTY, null,
				("EXCHANGE_NAME 1:::"+message).getBytes());
		channel.basicPublish(ChannelFactory.EXCHANGE_NAME_, StringUtils.EMPTY, null,
				("EXCHANGE_NAME 2:::"+message).getBytes());
		
		ChannelFactory.closeChannel(channel);
	}
}



consumer类,临时队列需要和两个exchange进行绑定:

class Logger {
	public static void main(String[] args) throws IOException,
			ShutdownSignalException, ConsumerCancelledException,
			InterruptedException {
		Channel channel = ChannelFactory.getChannel();

		String queue = channel.queueDeclare().getQueue();
		System.out.println("temporary queue name::"+queue);

		channel.queueBind(queue, ChannelFactory.EXCHANGE_NAME, "");
		channel.queueBind(queue, ChannelFactory.EXCHANGE_NAME_, "");

		QueueingConsumer consumer = new QueueingConsumer(channel);
		channel.basicConsume(queue, true, consumer);

		while (true) {
			System.out.println(new String(consumer.nextDelivery().getBody()));
		}
	}
}


由于使用的是临时队列,需要先运行consumer再运行producer。

运行结果输出:

bubuko.com,布布扣


本文出自 “Map.get(X)=new Object()” 博客,请务必保留此出处http://runtime.blog.51cto.com/7711135/1437297

RabbitMQ - Publish/Subscribe in Java,布布扣,bubuko.com

RabbitMQ - Publish/Subscribe in Java

标签:rabbitmq   java   publish   

原文地址:http://runtime.blog.51cto.com/7711135/1437297

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!