发布/订阅

在之前的教学中，我们创建了一个工作队列。工作队列是假设每个任务交付给一个工人。而在这部分我们将做一些完全不同的事儿，我们将一条信息分发给多个消费者。这种模式被叫做“发布/订阅”模式。

为了举例说明这种模式，我们将构建一个简单的日志记录系统。它将由两部分程序组成——第一部分是发送日志消息，第二部分将会接收并将其打印出来。

在我们的程序中，每个运行中的接收程序都将获得消息。这样我们就可以运行一个接收器并将日志写入磁盘；同时我们运行另一个可以在屏幕上看到日志。

基本情况是，发布的日志消息将被广播到所有的接收者上。

交换器

在本系列教程的前几部分，我们发送和接收了来自队列的消息。现在是时候来介绍Rabbit的完整消息模型了。

让我们快速的复习一下前面教程包含的内容：

生产者是发送消息的用户应用程序。

队列是存消息的缓冲区。

消费者是接收消息的用户应用程序。

RabbitMQ中的消息传递模型的核心思想是，生产者不会直接向队列发送任何消息。事实上，通常情况下，生产者甚至都不知道消息是否会被传送到任何队列。

其实，生产者只能发送消息给交换器。交换是件非常简单的工作。一方面，它接收来自生产者的消息，另一方面它将消息推送给队列。交换器必须知道如何处理它接收到的消息。那它是否需要关联到一个指定的队列？是否需要关联很多队列？或者说直接弃之不用。这些规则是由交换器类型定义的。

这里有几个交换器类型可供使用：direct、topic、headers和fanout。我们现在将关注最后一个——fanout。让我们创建一个这种模式的交换器，并给他命名为logs：                           

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

fanout交换非常简单。正如您可能从名称中猜测的那样，它只是将接收到的所有消息广播到它所知道的所有队列。这正是我们的日志系统所需要的。

交换器列表

  要在服务器上裂齿交换器，你可以运行特别有用的rabbitmqctl：

sudo rabbitmqctl list_exchanges

  在列表中会有一些amq.* 的交换器和一些默认（未命名）的交换器。这些是默认创建的，但现在你不太需要用到它们。

匿名交换器

  在前面的教学中，我们对交换器一无所知，但我们仍然可以发送消息到队列。那是因为我们使用了默认的交换器，这个默认交换器是由空字符串（“”）指定的。

   回想一下我们之前是如何发布消息的：

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

   第一个参数就是交换器的名字。空字符串表示默认或者是匿名的交换器。如果存在routingkey的话，消息将根据routingkey名称被路由到队列。（这里个人感觉有点绕，可以自己查询一下路由的概念，总之就是消息---->交换器---->队列）

现在，我们可以将其发布到我们命名的交换器中了：

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

临时队列

你可能还记得，我们使用的队列有指定的名称（还记得hello和task_queue么？）。可以给队列命名对我们来说是特别重要的，因为我们需要将worker指定到相同的队列。给队列起个名字是特别重要的，比如说当你想去通过同一个队列去在生产者和消费者之间传递消息。

但对于我们的日志系统情况并非如此。我们希望监听到所有的日志消息，而非其中的一个子集。并且，我们只想关心当前的消息，而不关心之前的。为了解决这些问题，我们需要做两件事儿。

首先，当我们连接得到Rabbit的时候，我们需要一个新的空队列。我们可以创建一个随机名称的队列，或者，更好的是让服务器给我们选择一个随机的队列名称。

其次，一旦消费者断开连接，就应该对应的删除队列。

在Java中，当我们不向queueDeclare()提供参数是，我们会创建一个非持久的，独立的，自动删除的队列，并且带有一个自动生成的名称：

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

在这里，queueName 是一个随机的队列名，例如，可能名字是“amq.gen-JzTY20BRgKO-HjmUJj0wLg.”

绑定（Bindings）

我们已经创建了一个fanout交换器和一个队列。现在我们需要告诉exchange将消息发送到我们的队列。交换器和队列之间的关系建立称为绑定。

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

这样，日志交换器就会将消息附加到我们的队列中了。

绑定列表

     你可以同过下面的命令列出现有的绑定：

rabbitmqctl list_bindings

整合代码

发布日志消息的生产者程序和之前的教程没有太大的不同。最重要的变化是，我们现在想要将消息发布到日志交换器中，而不是匿名交换器中。在发送时，我们需要提供一个routingKey（路由键），但对于fanout交换器，这个值会被忽略掉。下面是EmitLog.java程序：

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();

    }

    //...

}