注：这是RabbitMQ-java版Client的指导教程翻译系列文章，欢迎大家批评指正
第一篇Hello Word了解RabbitMQ的基本用法
第二篇Work Queues介绍队列的使用
第三篇Publish/Subscribe介绍转换器以及其中fanout类型
第四篇Routing介绍direct类型转换器
第五篇Topics介绍topic类型转换器
第六篇RPC介绍远程调用
发布/订阅（Publish/Subscribe）
在上篇指导教程中我们创建了一个工作队列。在一个工作队列中，每个任务都被精确的分配给一个工作者。但在这篇指导中，我们做一些不同的事情，就是把一条消息分发给多个消费者。这就是著名的“发布/订阅”模型。
为了阐释这个模型。我们打算建立一个日志系统，它有两个应用组成，一个将会发出日志消息，另一个将会接受并打印出日志消息。
在我们的日志系统中运行相同的消费者应用都会接受到消息，按照这种方式，我们可以运行一个接受者将日志消息持久到磁盘上，与此同时我们可以再运行一个接受者在屏幕上查看的日志消息。
本质上，被发布的日志消息会广播给所有的接受者。
转换器（Exchanges）
在上篇指导教程中我们发送和接受消息都是来自一个队列，而现在来介绍RabbitMQ完整消息模型。
我们快速的复习一下我们上篇指导教程中学习的东西：
1.一个生产者的应用可以发送消息
2.一个队列可以缓存许多消息
3.一个消费者的应用可以接受消息
而RabbitMQ中消息模型的核心思想是生产者不会直接发送消息给一个队列。实际上，经常是生产者都不知道生产的消息是会被分发到哪一个队列中。
实际上，生产者只会发送消息给转换器（exchanges）。转换器的原理非常简单，一方面从生产者接受消息，另一方面将消息推送到队列中。但转换器必须明确的知道接收到的消息要做什么。是被添加到一个特殊的队列中，还是被添加到多个队列中，亦或者被销毁。这些规则都是由转换器的类型决定的。
exchanges.png
这里有四种转换器的类型可使用：direct，topic，headers和fanout。我们先学习最后一个:fanout，创建一个fanout类型的转换器，叫着logs；
channel.exchangeDeclare("logs", "fanout"); //前者参数为名称，后者参数为转换器类型
fanout转换器非常简单，或许你从名字就可以猜出来，它把自己所有消息广播给绑定它的所有队列。这确切是我们日志所需要的。
匿名的转换器
在上篇指导教程中我们并不知道转换器，但是我们依然能都发送消息给队列。那是因为我们使用了一种默认的转换器，定义的一个空字符串。
回想以下我们是如何发布一条消息的：
channel.basicPublish("", "hello", null, message.getBytes());
第一个参数就是转换器的名字，空的字符串表示默认的或者说匿名的转换器：如果是routingkey存在的话，那么消息将会路由到指定routingkey的队列中。
现在，我们使用我们命名的转换器发送消息：
channel.basicPublish( "logs", "", null, message.getBytes());
临时的队列（Temporary queues）
可能你还记得以前我们使用队列的时候都会有一个具体的名字（像hello 和 task_queue），对我们来说命名一个队列实在太重要了，我们需要把多个工作这指向同一个队列。当你想在生产者和消费者之间共享一个队列，那给一个队列命名就太重要。
但是这个例子不适合我们的日志系统。我们想知道所有的日志消息，而并不是它们中的一部分。我们同样只想获取到当下的消息而不是以前的消息，为了解决这个问题我们需要确定两件事情。
第一，无论什么时候连接上RabbitMQ，我们有一个新的空的队列。为了这样，我们可以创建一个随机命名的队列，或者更好的方式是：让服务器为我们选择一个随机命名的队列。
第二，一旦消费者失去连接，这个队列会被自动删除。
在java的客户端，提供了一个没有参数的queueDeclare()方法能够创建一个临时，唯一，会自动删除且自动命名的队列：
String queueName = channel.queueDeclare().getQueue();
这样的话，这个队列就是一个随机命名的。举例来说它有点像：amq.gen-JzTY20BRgKO-HjmUJj0wLg
绑定（Bindings）
bindings.png
我们已经创建了一个fanout转换器和一个队列。现在我们需要告诉转换器发送消息给我们的队列。转换器和队列之间的关系就叫着绑定(binding):
channel.queueBind(queueName, "logs", "");
从现在开始，logs转换器将会添加消息到队列。
展示所有的绑定
可以展示已经存在的绑定关系，命令如下：
rabbitmqctl list_bindings
综合
python-three-overall.png
发送消息的生产者应用，跟之前指导教程中的相差并不大。但是，最大的不同是我们现在需要将消息发布给名叫logs的转换器，而不是匿名的转换器。发送消息时本应该提供一个routingkey的值，但是对于fanout类型的转换器，routingkey属性就没有意义。
下面是EmitLog.java类的源代码，这里下载：
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();
}
//...
}
正如你所看到的，建立连接后，就声明了转换器，这一步是非常重要的。如果发布到一个不存在的转换器上是被禁止的。（预计会报错吧）
如果没有队列绑定这个转换器，消息将会被丢失。但对我们来说是没问题的，如果没有消费者监听，那我们可以安全的删除这个消息。
下面是ReceiveLogs.java类的源代码，这里下载：
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);
}
}
先编译，再完成：
javac -cp $CP EmitLog.java ReceiveLogs.java
如果你想保留日志到文件中，只需要打开一个窗口和记录：
java -cp $CP ReceiveLogs > logs_from_rabbit.log
如果你想在你的屏幕上看到日志，运行一个终端：
java -cp $CP ReceiveLogs
当然，发送日志类型：
java -cp $CP EmitLog
使用rabbitmqctl list_bindings命令行，你可以核实是否是我们想要的绑定和队列。有两个ReceiveLogs.java应用在运行，应该就可以看到下面的情况：
sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
# => logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
# => ...done.
结果是明确的：数据从logs转换器发送至两个服务端命名的队列中，这确切来说就是我们的意图。
第三节的内容大致翻译完了,这里是原文链接。接着进入下一节：Routing。
终篇是我对RabbitMQ使用理解的总结文章，欢迎讨教。
--谢谢--
网友评论