rabbitmq官方教程之发布与订阅（Publish/Subscribe）

傷城~ 2022-06-04 00:35 198阅读 0赞

**官网教程地址：[http://www.rabbitmq.com/tutorials/tutorial-one-java.html][http_www.rabbitmq.com_tutorials_tutorial-one-java.html]**  
Publish/Subscribe

(using the Java Client)

在上一篇工作队列中中我们已经创建了一个工作队列。工作队列背后的假设是每个任务都交付给一个worker。

在这部分中，我们会做一些完全不同的事情 - 我们会向多个消费者传递信息。这种模式被称为“发布/订阅”。

为了说明这个模式，我们要建立一个简单的发布订阅系统。它将包括两个程序 - 第一个将发出消息，第二个将接收并打印它们。

在这个日志系统中，每一个接收程序（消费者）都会收到所有的消息，其中一个消费者将消息直接保存到磁盘中，而另一个消费者则将日志输出到控制台。

从本质上讲，发布的日志消息将会广播给所有的接收者（消费者）。

**Exchanges**

在之前的教程里，我们都是直接往队列里发送消息，然后又直接从队列里取出消息。现在是时候介绍RabbitMQ的整个消息模型了。

先让我们快速地回顾一下之前教程中的几个概念：

*  生产者：发送消息的用户程序
 *  队列：存储消息的缓冲区
 *  消费者：接收消息的用户程序

RabbitMQ的消息模型中的一个核心思想是，生产者绝不会将消息直接发送到队列中，实际上，在大部分场景中生产者根本不知道消息会发送到哪些队列中。

恰恰相反，生产者只能将信息发送到exchanges。exchanges做的事情也很简单。一方面，它收到来自生产者的消息，另一方将它们推送到队列。但是Exchange必须清楚地知道怎么处理接收到的消息：是将消息放到一个特定的队列中，还是放到多个队列中，还是直接将消息丢弃，下图示意了Exchange在消息模型中的位置：  
  
![Exchange在消息模型中的位置][Exchange]  
**注：X指Exchange**  
  
Exchange一共有四种类型：direct、topic、headers 和fanout。  
  
此教程将会使用fanout类型的Exchange，让我们创建一个名为logs的fanout类型的Exchange

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

fanout类型的Exchange非常简单，从它的名字你可能就已经猜出来了（说实话我真没猜出来！！！！我只知道fanout是扇形的意思），**它将会将接收到的消息广播给所有它知道的队列**。这正是我们的日志系统所需要的类型。  
  
【注这句话将是重点：**它将会将接收到的消息广播给所有它知道的队列】**  
  
  
  
可以通过下面的命令列出Rabbit服务器上的所有Exchange

sudo rabbitmqctl list_exchanges

注：我试了其实只需要切换到sbin目录下，然后

rabbitmqctl list_exchanges

### 默认exchange（原文是Nameless exchange）       ###

##### 在前面的教程中，我们对Exchange一无所知，但是我们仍然可以将消息发送到队列中，      这可能是因为我们使用了默认的Exchange，我们是通过空字符串”“来定义这个Exchange的。      回顾一下我们之前是怎么发布消息的： #####

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

上面代码的方法中，第一个参数就是Exchange的名字，空字符串表示默认或无名Exchange：消息通过由routingKey定义的队列被路由的。  
  
现在，我们通过下面的方式来发布消息：

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

### 临时队列(Temporary queues)    ###

##### 你可能记得之前我们使用了特定名字的队列（还记得hello和task\_queue吗）。      可以指明一个队列这一点对我们而言至关重要，因为我们也要让工作者指向同一个队列。      当你在生产者和消费者之间共用一个队列时，给这个队列取个名字就非常重要。      但这不适用于我们的日志系统。我们想让每个消费者都接收到所有的日志消息，      而不是其中的一部分日志消息。我们关心的是当前广播的消息而不是之前的那些。      为了解决这些问题，我们需要做两件事情。 #####

首先，不管怎样我们连接到RabbitMQ服务的时候，我们都需要一个新的空队列。  
  
为了达到这个效果，我们可以为队列取一个随机的名字，或者更好的是，  
  
让RabbitMQ服务器为我们的队列随机起个名字。

第二件事：当我们关闭了消费者的时候，队列应该自动删除。

当我们调用无参的queueDeclare()的时候，意味着创建了一个非持久、独特的、自动删除的队列，并返回一个自动生成的名字：

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

这样就可以获取随机的队列名字了，这个名字看起来像：`amq.gen-JzTY20BRgKO-HjmUJj0wLg.`

### 绑定(Bindings)   ![bindings][]       ###

##### 我们已经创建了一个fanout类型的Exchange和一个队列。      现在我们需要告诉Exchange发送消息到我们的队列中。      Exchange和队列之间的关系称为绑定。 #####

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

这样，我们创建的队列就和我们创建的logs路由器建立了关系，路由器就会将消息发送到这个队列中。

可以通过如下的命令查看所有已经存在的绑定关系：

rabbitmqctl list_bindings

#### 把他们放一起（Putting it all together）   ![这里写图片描述][python-three-overall.png]    ####

##### 对生产者程序，它输出日志消息，与之前的教程并没与很大不同， #####

#####  #####

##### 最大的不同在于我们将消息发布给logs路由器，而不是无名的exchange ， #####

#####  #####

##### 而参数中的routingKey由于发布模式是fanout所有可以用空值代替，以下是生产者 #####

#####  #####

##### EmitLog.java的完整代码： #####

import com.rabbitmq.client.BuiltinExchangeType;
    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 Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
         //声明路由以及路由的类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    
        String message = getMessage(argv);
        //发布消息
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");
    
        channel.close();
        connection.close();
      }
    
      private static String getMessage(String[] strings){
        if (strings.length < 1)
                return "info: Hello World!";
        return joinStrings(strings, " ");
      }
    
      private static String joinStrings(String[] strings, String delimiter) {
        int length = strings.length;
        if (length == 0) return "";
        StringBuilder words = new StringBuilder(strings[0]);
        for (int i = 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
      }
    }

可以看到，在建立了连接之后，我们声明了路由器Exchange。这一步是必须的，因为不允许将消息发给一个不存在的路由器。

如果路由器还没有绑定队列，这些发送给路由器的消息将会丢失。但这对我们影响不大，如果还没有消费者监听，我们可以安全地丢弃这些消息。

消费者完整代码：The code for 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, BuiltinExchangeType.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);
      }
    }

后边的就不翻译了，就是测试，查看结果；先运行两个消费者，运行生产者EmitLog.java

为了验证我们的代码真正地将队列和路由器绑定到了一起，可以使用rabbitmqctl list\_bindings命令查看绑定关系，假定我们运行了两个消费者，那么你应该可以看到如下的类似信息：

sudo rabbitmqctl list_bindings
    # => Listing bindings ...
    # => logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
    # => logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
    # => ...done.

从上面的结果可以看到，数据从logs路由器传输到两个随机名字的队列中，这正是我们想要的。

想要了解如何监听一部分消息，请查看[教程4][4]。

[http_www.rabbitmq.com_tutorials_tutorial-one-java.html]: http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[Exchange]: http://www.rabbitmq.com/img/tutorials/exchanges.png
[bindings]: http://www.rabbitmq.com/img/tutorials/bindings.png
[python-three-overall.png]: http://www.rabbitmq.com/img/tutorials/python-three-overall.png
[4]: http://www.rabbitmq.com/tutorials/tutorial-four-java.html