RabbitMQ之发布订阅【译】

lilingjie2015

　　在上一节中我们创建了一个工作队列，最好的情况是工作队列能够把任务恰到好处的分配给每一个worker。这一节中我们将做一些完全不同的事情——将消息传递给每一个消费者，这种模式被称为发布/订阅。
　　
为了说明这种模式，我们构建一个日志系统，其包括两个部分，一个发出日志消息，另一个接收并打印出来。
　　
在我们的日志系统里，每一个拷贝的接收者都将收到消息，这样我们就能够跑一个接收者将收到的日志消息写入磁盘，并在同时另外一个接收者将日志消息打印在屏幕上。
　　
本质上来说，发布日志消息将要被广播到所有的接收器。

交换器（Exchanges）
　　在本教程之前我们在队列中发送和接收消息，现在我们将介绍Rabbit中完整的消息传递模式。
　　
我们先回顾一下前面的教程：

　　生产者发送消息
　　
队列是一个缓冲区存储消息
　　
消费者接收消息

　　RabbitMQ消息模式的核心是生产者从来不会直接发送一个消息给队列，事实上，发送者都不知道消息将要被发送到一个队列。
　　
相反，生产者发送消息到exchange，exchange是一个非常简单的东西，它一方面接收生产者的消息，另一方面推送消息到队列，exchange必须知道如何处理它接收到的消息，是应该将消息附加至一个特定的队列？还是附加到很多队列？或者而应该丢弃？这些规则由exchange来决定。
　　

　　
exchange包含以下几种类型：direct, topic, headers 和 fanout。我们将重点放在最后一个——fanout，我们传建一个这种类型的exchange，命名为logs：
　　

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

　　fanout类型的exchange非常简单，我们可以从名字就猜出来，它只是广播接收到的所有消息至所有的队列。而这正是我们的日志系统所需要的。

　　列出exchange
　　
我们在服务器上将所有的exchange列出来
　　
$ sudo rabbitmqctl list_exchanges
　　
Listing exchanges ...
　　
direct
　　
amq.direct direct
　　
amq.fanout fanout
　　
amq.headers headers
　　
amq.match headers
　　
amq.rabbitmq.log topic
　　
amq.rabbitmq.trace topic
　　
amq.topic topic
　　
logs fanout
　　
...done.
　　
在这些list里面我们可看到一些amq.* exchanges，这些是默认的exchange。
　　无名的exchange
　　
在上一节中，我们对于exchange什么都不知道，但是我们依旧可以发送消息至队列，这是因为我们使用了使用了默认的exchange，它被定义为一个空字符串“”。
　　
channel.basicPublish("", "hello", null, message.getBytes());
　　
第一个参数就是exchange，空字符串表示默认或者无名交换。如果routingkey存在的话，消息会根据routingkey被路由到队列。

　　现在我们发布到有名的exchange：
　　

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

临时队列（Temporary queues）
　　我们在之前的教程中给队列起了名字（helll和task_queue）,给一个队列命名对于我们来说是很重要的，我们需要消费者指向同一个队列。
　　
但是这不是我们目前的日志程序所需要的，我们希望能接受所有的消息，而不是这些消息的子集。当然我们也只是关注当前流动的消息，而不需要关注那些过期的消息。这样，我们需要做两件事情。
　　
第一，无论何时我们连接Rabbit，我们需要一个新的空的队列，这样需要用随机的名字来创建一个队列，更好地，我们让服务器来随机的生成一个队列名字。
　　
第二，我们一旦和消费者断开，队列必须马上删除。
　　
在Java Client中，我们不提供参数给queueDeclare() 方法，就能生成一个非持久，独特，自动删除队列的名称：
　　

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

　　这样队列拥有一个随机的队列名称，比如：amq.gen-JzTY20BRgKO-HjmUJj0wLg。

绑定（Bindings）
　　现在我们已经建立了一个exchange和一个队列，现在我们需要告诉exchange将消息发送到我们的队列中，exchange和队列之间的关系我们成为Binding。
　　

　　

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

　　从现在开始日志程序中的exchange就可以将消息附加到队列上了。

　　列出binding的列表
　　
你可以列出现有的Binding列表，命令是：rabbitmqctl list_bindings.

代码整合
　　生产者和之前的教程中相差不多，最大的改变是我们现在需要将消息发布到logs exchange，而不是之前的未命名exchange，当发送消息的时候我们需要提供一个routingKey，但是在fanout的消息模式中被忽略了。
　　

　　EmitLog.java：
　　

import java.io.IOException;　　
import com.rabbitmq.client.ConnectionFactory;
　　
import com.rabbitmq.client.Connection;
　　
import com.rabbitmq.client.Channel;
　　


　　
public>　　

　　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，因为发布到一个不存在的exchange是被禁止的。
　　
如果没有队列绑定到exchange，信息会丢失，但是也没关系。如果没有消费者在监听，我们可以安全的丢弃消息。
　　ReceiveLogs.java：
　　

import com.rabbitmq.client.*;　　

　　
import java.io.IOException;
　　


　　
public>　　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);
　　}
　　
}
　　

　　原文地址：https://www.rabbitmq.com/tutorials/tutorial-three-java.html
　　代码地址：https://github.com/aheizi/hi-mq
　　相关：
　　
1.RabbitMQ之HelloWorld
　　
2.RabbitMQ之任务队列
　　
3.RabbitMQ之发布订阅
　　
4.RabbitMQ之路由(Routing)
　　
5.RabbitMQ之主题(Topic)
　　
6.RabbitMQ之远程过程调用(RPC)