RabbitMQ 基础教程PublishSubscribe（发布订阅）

上篇文章中，我们把每个Message都是deliver到某个Consumer。在这篇文章中，我们将会将同一个Message deliver到多个Consumer中。这个模式也被成为 “publish / subscribe”。

这篇文章中，我们将创建一个日志系统，它包含两个部分：第一个部分是发出log（Producer），第二个部分接收到并打印（Consumer）。 我们将构建两个Consumer，第一个将log写到物理磁盘上；第二个将log输出的屏幕。

一、Exchanges

RabbitMQ 的Messaging Model就是Producer并不会直接发送Message到queue。实际上，Producer并不知道它发送的Message是否已经到达queue。

Producer发送的Message实际上是发到了Exchange中。它的功能也很简单：从Producer接收Message，然后投递到queue中。Exchange需要知道如何处理Message，是把它放到那个queue中，还是放到多个queue中？这个rule是通过Exchange 的类型定义的。 我们知道有三种类型的Exchange：direct, topic 和fanout。fanout就是广播模式，会将所有的Message都放到它所知道的queue中。创建一个名字为logs，类型为fanout的Exchange：

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

Listing exchanges

通过rabbitmqctl可以列出当前所有的Exchange：

sudo rabbitmqctl list_exchanges

注意 amq.* exchanges 和the default (unnamed)exchange是RabbitMQ默认创建的。

默认exchanges

上几节我们通过“”来指定默认的exchanges帮助我们分发消息给queue，像下面这样：

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

默认的分发机制是通过routingKey来匹配queue 现在我们可以通过exchange，而不是routing_key来publish Message了：

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

二、Temporary queues（临时queues）

截至现在，我们用的queue都是有名字的：第一个是hello，第二个是task_queue。使用有名字的queue，使得在Producer和Consumer之前共享queue成为可能。 但是对于我们将要构建的日志系统，并不需要有名字的queue。我们希望得到所有的log，而不是它们中间的一部分。而且我们只对当前的log感兴趣。为了实现这个目标，我们需要两件事情：

每当Consumer连接时，我们需要一个新的，空的queue。因为我们不对老的log感兴趣。幸运的是，如果在声明queue时不指定名字，那么RabbitMQ会随机为我们选择这个名字。方法：当Consumer关闭连接时，这个queue要被deleted。 String queueName = channel.queueDeclare().getQueue();

这样我们就能得到一个新的，自动删除的，随机分配名字的queue

三、Bindings（绑定）

现在我们已经创建了fanout类型的exchange和没有名字的queue（实际上是RabbitMQ帮我们取了名字）。那exchange怎么样知道它的Message发送到哪个queue呢？答案就是通过bindings：绑定。

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

现在logs的exchange就将它的Message附加到我们创建的queue了

Listing bindings

你可以通过以下命令来获取当前已经存在的binding

rabbitmqctl list_bindings

四、最终版本

Producer，在这里就是产生log的program，基本上和前几个都差不多。最主要的区别就是publish通过了exchange而不是routing_key。

import java.io.IOException; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; public class EmitLog { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws java.io.IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String message = getMessage(argv); channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } //... }

还有一点要注意的是我们声明了exchange。publish到一个不存在的exchange是被禁止的。如果没有queue bindings exchange的话，log是被丢弃的。

import com.rabbitmq.client.*; import java.io.IOException; public class ReceiveLogs { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + message + "'"); } }; channel.basicConsume(queueName, true, consumer); } }