RabbitMQ（三）：Exchange交换器--fanout

桃扇骨 2022-04-24 19:38 270阅读 0赞

内容翻译自：[RabbitMQ Tutorials Java版][RabbitMQ Tutorials Java]

--------------------

[RabbitMQ（一）：Hello World程序][RabbitMQ_Hello World]

[RabbitMQ（二）：Work Queues、循环分发、消息确认、持久化、公平分发][RabbitMQ_Work Queues]

[RabbitMQ（三）：Exchange交换器--fanout][RabbitMQ_Exchange_--fanout]

[RabbitMQ（四）：Exchange交换器--direct][RabbitMQ_Exchange_--direct]

[RabbitMQ（五）：Exchange交换器--topic][RabbitMQ_Exchange_--topic]

[RabbitMQ（六）：回调队列callback queue、关联标识correlation id、实现简单的RPC系统][RabbitMQ_callback queue_correlation id_RPC]

[RabbitMQ（七）：常用方法说明 与 学习小结][RabbitMQ_ _]

--------------------

### Publish/Subscribe： ###

在上一篇博客中，我们创建了一个工作队列：一个消息只能发送到一个工作者（消费者）中。而在这个教程中我们将会做完全不同的事情：我们发送同一个消息到多个消费者中。这种模式一般被称为“发布/订阅”模式。

为了演示这种模式，我们将会创建一个简单的日志系统。它由两个程序组成：第一个将会输出日志消息，第二个将会接受并打印出日志消息。

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

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

--------------------

### 交换器Exchanges： ###

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

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

（1）生产者：发送消息的用户程序

（2）队列：存储消息的缓冲区

（3）消费者：接收消息的用户程序

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

相反，生产者只会将消息发送给一个Exchange（路由器/交换器）。Exchange其实很简单，它所做的就是，接收生产者发来的消息，并将这些消息推送到队列中。Exchange必须清楚地知道怎么处理接收到的消息：是将消息放到一个特定的队列中，还是放到多个队列中，还是直接将消息丢弃。下图示意了Exchange在消息模型中的位置：

![20210324003641711.png][]

Exchange一共有四种类型：`direct`、`topic`、`headers` 和`fanout`。今天的教程将会使用`fanout`类型的Exchange，让我们创建一个名为`logs`的`fanout`类型的Exchange

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

`fanout`类型的Exchange非常简单，从它的名字你可能就已经猜出来了（fanout翻译过来是扇形的意思），它将会将接收到的消息广播给所有它知道的队列。这正是我们的日志系统所需要的类型。

> 可以通过下面的命令列出Rabbit服务器上的所有Exchange

sudo rabbitmqctl list_exchanges

没有命名的Exchange：

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

channel.basicPublish("", "hello", null, message.getBytes());
    //该方法的定义为：
    basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)

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

现在，我们通过下面的方式来发布消息：

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

--------------------

### 临时队列： ###

你可能记得之前我们使用了特定名字的队列（还记得`hello`和`task_queue`吗）。可以指明一个队列这一点对我们而言至关重要，因为我们也要让工作者指向同一个队列。当你在生产者和消费者之间共用一个队列时，给这个队列取个名字就非常重要。

但这不适应于我们的日志系统。我们想让每个消费者都接收到所有的日志消息，而不是其中的一部分日志消息。我们关心的是当前广播的消息。为了解决这些问题，我们需要做两件事情。

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

其次，当我们关闭了消费者的时候，队列应该自动删除。

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

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

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

--------------------

### 绑定： ###

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

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

![20210324003846972.png][]

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

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

rabbitmqctl list_bindings

--------------------

### 整合到一起： ###

![20210324003900510.png][]

对生产者程序，它输出日志消息，与之前的教程并没与很大不同。最重要的改变就是，我们将消息发布给`logs`路由器，而不是无名的路由的。当发消息的时候，我们需要提供一个路由键`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 = "msg...";
    
            //发布消息
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
    
            //关闭连接和通道
            channel.close();
            connection.close();
        }
    }

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

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

消费者`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);
        }
    }

现在，可以运行程序并查看结果了。首先运行两个消费者实例`ReceiveLogs.java`，然后运行生产者`EmitLog.java`。看看两个消费者实例是不是都接收到了所有的消息。

//生产者
     [x] Sent 'msg...'
    //消费者1
     [*] Waiting for messages. To exit press CTRL+C
     [x] Received 'msg...'
    //消费者2
     [*] Waiting for messages. To exit press CTRL+C
     [x] Received 'msg...'

可以看到，当生产者发出消息后，两个消费者最终都收到了消息。（本例子中只是都将日志消息都输出到控制台，如果想让其中一个消费者把日志消息输出到文件中，请参考原文）。

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

Listing bindings ...
    ...
    logs    exchange  amq.gen-S2B2k3mSnoLNNyUppS98Vw    queue  []
    logs    exchange  amq.gen-ljp5AngXol4iPW649OY7Pw    queue  []

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

想要了解如何监听一部分消息，请看下一篇博客。

--------------------

## ##

### 说明： ###

①与原文略有出入，如有疑问，请参阅原文

②原文均是编译后通过`javacp`命令直接运行程序，我是在IDE中进行的，相应的操作做了修改。

[RabbitMQ Tutorials Java]: https://link.jianshu.com?t=http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[RabbitMQ_Hello World]: https://blog.csdn.net/a745233700/article/details/89680776
[RabbitMQ_Work Queues]: https://blog.csdn.net/a745233700/article/details/89680957
[RabbitMQ_Exchange_--fanout]: https://blog.csdn.net/a745233700/article/details/89681388
[RabbitMQ_Exchange_--direct]: https://blog.csdn.net/a745233700/article/details/89681621
[RabbitMQ_Exchange_--topic]: https://blog.csdn.net/a745233700/article/details/89681801
[RabbitMQ_callback queue_correlation id_RPC]: https://blog.csdn.net/a745233700/article/details/89681941
[RabbitMQ_ _]: https://blog.csdn.net/a745233700/article/details/89703132
[20210324003641711.png]: /images/20220212/a2514465e7a04524925012044e8aa3ab.png
[20210324003846972.png]: /images/20220212/9b3b6f60d6164d8a8e04114f40a9d0ba.png
[20210324003900510.png]: /images/20220212/ec8c6dc9282d4da2bc568b77593d6a75.png