RabbitMQ入门学习——Publish/Subscribe 发布/订阅

 

http://www.rabbitmq.com/tutorials/tutorial-three-java.html

                   
 

在前面的章节中，我们创建了一个工作队列。工作队列假设每一个任务都交给一个工作者来处理。在这一章节，我们会处理一些完全不同的事情，我们会将消息发送给多个消费者。这种模式就叫做“publish/subscribe”

为了说明这种模式，我们会建立一个简单的日志系统。它包括两个程序：一个发送日志，另一个接受然后将他们打印出来。

在我们的日志系统中，每一个运行这的接受者，都将会收到消息。通过这种方式，我们可以在同一时间使一个接受者将消息保存到硬盘上，另一个接受者将消息打印到控制台。

本质上，被发布的消息将会广播到每一个接受者上

1：Exchanges(交换区)

在前面的章节中，我们都是将消息写到队列中，然后从队列中取出消息。现在是时候引入完整的Rabbit消息模型了

让我们回顾下前面章节都包括那些内容

一个producer发送消息的应用

一个queue缓存消息

一个consumer接受消息的应用

RabbitMQ消息模型的核心思想就是，生产者不把消息直接发送给队列。实际上，生产者在很多情况下都不知道消息是否会被发送到一个队列中。取而代之的是，生产者将消息发送到交换区。交换区是一个非常简单的东西，它一端接受生产者的消息，另一端将他们推送到队列中。交换区必须要明确的指导如何处理它接受到的消息。是放到一个队列中，还是放到多个队列中，亦或是被丢弃。这些规则可以通过交换区的类型来定义。

可用的交换区类型有：direct，topic，headers，fanout。下面将集中讨论最后一个：fanout。可以通过channel.exchangeDeclare(“logs”,”fanout”)来创建一个这种类型的交换区。它会把它接收到的所有消息，发送给所有和它绑定的队列。并且，这正是我们日志需要的。

可以通过客户端的rabbitmqctl list_exchanges查看所有的交换区可以通过rabbitmqctl list_exchanges列出所有的交换区。amp.*表示的是默认或没有命名的交换区。在前面的章节中，我们不知道交换区，但是我们还是可以发送消息到队列中。那是因为我们使用了默认的交换区，我们通过控制符串来制定。回想之前发布消息的代码channel.basicPublish(“”,”hello”,null,message.getBytes());第一个参数就是交换区的名字。空字符串表示这是默认的或者没有命名的交换区。消息通过指定的路由名routingkey被路由到队列中。

    现在我们可以将消息发送到交换区

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

 

2：Temporary queues 临时队列

之前使用的队列中有一个名字，给定一个名字对我们来说非常的重要，因为我们需要将消费者关联到队列。当我们希望在生产者和消费者之间共享队列时，指定队列的名字将变的非常重要。

但是，这不适用于我们的日志记录。我们希望记录所有的日志消息，而不是他们的一个子集。我们同时也对当前的信息感兴趣而不是老得信息。要解决这些问题，我们需要做如下两件事情：

首先：当我们连接到RabbitMQ时，我们需要一个新的，空的队列。为了解决这个问题，我们可以使用一个随机创建的名字；或者让RabbitMQ服务器随机帮我们创建一个队列名字。

其次：一旦我们断开连接，需要自动删除消费者队列

我们可以通过channel.queueDeclare()来创建一个非持久的队列名，并通过getQueue()来返回对列名。

3: Bindings(绑定)

我们已经创建了一个fanout交换区和一个队列，现在我们就可以让交换区来发送消息到我们的队列中来。交换区和队列的关系，可以通过调用channel.queueBind(queueName,”logs”,””);来绑定。

现在，所有的日志交换区将会将数据追加到我们的队列中

通过客户端Rabbitmqctl list_bindings命令，可以查看所有的绑定

4：Putting it all together

发送消息的生产者程序，和之前的代码看起来没有多大的区别。最主要的变化就是，我们现在希望将消息发送给我们的logs交换区，而不是之前的默认交换区。发送时，我们需要提供一个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();

    }

    //...

}

正如你所看到的，在建立到RabbitMQ的连接之后，我们就申明了一个交换区。这一步非常重要，因为发布消息到一个不存在的交换区是不被允许的。如果没有队列绑定到交换区，消息将会丢失。

ReceiveLogs.java

import java.io.IOException;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.QueueingConsumer;

 

public class ReceiveLogs {

 

    private static final String EXCHANGE_NAME = "logs";

 

    public static void main(String[] argv)

                  throws java.io.IOException,

                  java.lang.InterruptedException {

 

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

 

        QueueingConsumer consumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName, true, consumer);

 

        while (true) {

            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            String message = new String(delivery.getBody());

 

            System.out.println(" [x] Received '" + message + "'");

        }

    }

}

使用rabbitmqctl list_bindings命令，你可以确认程序却是创建了bindings和queue。