100.交换机

在上一节中，我们创建了一个工作队列。我们假设的是工作队列背后，每个任务都恰好交付给一个消费者（工作进程）。在这一部分中，我们将做一些完全不同的事情-我们将消息传达给多个消费者，这种模式称为“发布/订阅”

为了说明这种模式，我们将构建一个简单的日志系统。它将由两个程序组成：第一个程序将发出日志消息，第二个程序是消费者。其中我们会启动两个消费者，其中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把消息打印在屏幕上，也就是第一个程序发出的日志消息将广播给所有消费者

在讲发布/订阅之前，我们先讲讲交换机

Exchanges

RabbitMQ 消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。

生产者只能将消息发送到交换机（exchange），交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。

交换机必须确切知道如何处理收到的消息：是应该把这些消息放到特定队列，还是说把他们到许多队列中，还是说应该丢弃它们。这就的由交换机的类型来决定，交换机有这些类型：默认类型（无名类型），直接（direct），主题（topic） ，标题（headers），扇出（fanout，也叫发布订阅）

‍

‍

无名 exchange

在本教程的前面部分我们对 exchange 一无所知，但仍然能够将消息发送到队列。之前能实现的原因是因为我们使用的是默认交换机，我们通过空字符串进行标识：

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

第一个参数是交换机的名称。空字符串表示默认或无名称交换机。消息能路由发送到队列中其实是由 routingKey(bindingkey) 绑定 key 指定的（如果有该参数，没有则使用默认）

‍

也可以在后台看到有一个默认交换机：

‍

‍

而使用默认交换机，我们是无法实现一个消息发给多个消费者的，还记得我们之前说过两个消费者之间是竞争关系吗？一个消息被消费后，另一个消费者就不能拿到该消息了。

‍

‍

而我们可以用一个交换机，通过路由规则绑定 2 个队列，然后一个消息也放到两个队列上，这样虽然队列中的消息只能被消费一次，但我们有多个队列，这样一个消息就能被消费多次了。

‍

‍

临时队列

之前的章节我们使用的是具有特定名称的队列（还记得 hello 和 ack_queue 吗？）。队列的名称对我们来说至关重要，我们需要指定我们的消费者去消费哪个队列的消息。

‍

每当我们连接到 Rabbit 时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。因此我们可以用临时队列

‍

创建临时队列的方式如下:

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

‍

‍

创建出来之后长成这样：

‍

‍

‍

绑定（bindings）

binding 是 exchange 和 queue 之间的桥梁，它告诉我们 exchange 和哪个队列进行了绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定

‍

‍

我们可以演示下。首先新建一个队列：

‍

同理，新建一个交换机

‍

‍

然后我们点击新建的交换机，并添加 bindings：binding key 随便写一个

‍

‍

然后就可以看到有绑定关系了：

‍

‍

一旦发消息到该交换机，交换机就可以根据路由规则，发送到对应的队列，队列再将队列发给消费者。为什么要有 routing key？这是因为有可能生产者只想发给某一个队列，此时就可以指定 routing key。

此时发消息就变的很灵活。

‍

‍

‍

Fanout（扇出）

Fanout 这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些 exchange 类型，其中就包含扇出

‍

下面我们来实践下，将一个交换机（取名为 logs）绑定 2 个队列，发送一个消息，然后两个队列都能收到并被消费，示意图：

‍

生产者

package com.peterjxl.rabbitmq.demo6;


import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.Channel;

public class EmitLog {
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        for (int i = 0; i < 10; i++) {
            String message = "Hello World " + i;
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println("生产者发送消息: " + message);
        }
    }
}

‍

消费者 01

package com.peterjxl.rabbitmq.demo6;

import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.Channel;

public class ReceiveLogs01 {

    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 生成一个临时队列, 队列名称是随机的, 当消费者断开与队列的连接时, 队列自动删除,
        String queueName = channel.queueDeclare().getQueue();

        // 将临时队列绑定到交换机上, 参数3是routingKey, 交换机根据这个key将消息转发到指定的队列上
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        System.out.println("ReceiveLogs01等待接收消息,把接收到的消息打印在屏幕上......");

        // 接收消息, 参数2是自动确认, 参数3是消息的回调
        channel.basicConsume(queueName, true, (consumerTag, message) -> {
            System.out.println("ReceiveLogs01接收到的消息: " + new String(message.getBody()));
        }, consumerTag -> {
        });
    }
}

‍

‍

关键在于第 20 行的绑定：

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

这里空字符串表示不指定，也就是交换机会发送消息给所有被绑定的队列。

‍

‍

消费者 02

package com.peterjxl.rabbitmq.demo6;

import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.Channel;

public class ReceiveLogs02 {
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 生成一个临时队列, 队列名称是随机的, 当消费者断开与队列的连接时, 队列自动删除,
        String queueName = channel.queueDeclare().getQueue();

        // 将临时队列绑定到交换机上, 参数3是routingKey, 交换机根据这个key将消息转发到指定的队列上
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        System.out.println("ReceiveLogs02等待接收消息,把接收到的消息打印在屏幕上......");

        // 接收消息, 参数2是自动确认, 参数3是消息的回调
        channel.basicConsume(queueName, true, (consumerTag, message) -> {
            System.out.println("ReceiveLogs02接收到的消息: " + new String(message.getBody()));
        }, consumerTag -> {
        });
    }
}

‍

测试

我们先启动两个消费者，然后再启动生产者，可以看到两个消费者都接受到了消息：

‍

在控制台也能看到有这个 exchange：

‍

Direct exchange

在上一节中，我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。在本节我们将向其中添加一些特别的功能：比方说我们只让某个消费者订阅发布的部分消息。例如我们只把严重错误消息定向存储到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

Fanout 这种交换类型并不能给我们带来很大的灵活性，它只能进行无意识的广播，在这里我们将使用 direct 这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的 routingKey 队列中去。

我们再次来回顾一下什么是 bindings，绑定是交换机和队列之间的桥梁关系。也可以这么理解：队列只对它绑定的交换机的消息感兴趣。绑定用参数用 routingKey 来表示。也可称该参数为 binding key，创建绑定：

channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");

‍

绑定之后的意义由其交换类型决定。举个例子：

‍

在上面这张图中，我们可以看到 X 绑定了两个队列，绑定类型是 direct。队列 Q1 绑定键为 orange，队列 Q2 绑定键有两个：一个绑定键为 black，另一个绑定键为 green.

在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 black 或 green 和的消息会被发布到队列 Q2，其他消息类型的消息将被丢弃。

和扇出有点类似，扇出的时候两个交换机和队列的 routing key 是相同的（空字符串），因此两个队列都能收到。而直接交换机，就有点像是扇出交换机的变种，因为 routing key 不同，所以不会广播给所有队列。

如果 exchange 的绑定类型是 direct，但是它绑定的多个队列的 key 如果都相同，在这种情况下虽然绑定类型是 direct ，但是它表现的就和 fanout 有点类似了，就跟广播差不多，如下图所示，这种情况也叫多重绑定

‍

接下来我们来实践下

‍

需求

定义一个交换机和 2 个队列，并根据日志级别绑定不同的 routing key：

‍

‍

绑定关系：有 2 个队列，但有 3 个绑定关系

‍

‍

消费者 01

写一个消费者，声明一个交换机和队列，并绑定两个 routing key

package com.peterjxl.rabbitmq.demo7;

import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class ReceiveLogsDirect01 {
    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        channel.queueDeclare("console", false, false, false, null);
        channel.queueBind("console", EXCHANGE_NAME, "info");
        channel.queueBind("console", EXCHANGE_NAME, "warning");

        System.out.println("ReceiveLogsDirect01等待接收消息，把接收到的消息打印在屏幕上......");
        channel.basicConsume("console", true, (consumerTag, message) -> {
            System.out.println("ReceiveLogsDirect01控制台打印接收到的消息：" + new String(message.getBody()));
        }, consumerTag -> {});
    }
}

‍

‍

消费者 02

和上一个消费者代码类似，声明一个交换机和队列，并绑定 routing key

package com.peterjxl.rabbitmq.demo7;

import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class ReceiveLogsDirect02 {
    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        channel.queueDeclare("disk", false, false, false, null);
        channel.queueBind("disk", EXCHANGE_NAME, "error");

        System.out.println("ReceiveLogsDirect02等待接收消息，把接收到的消息打印在屏幕上......");
        channel.basicConsume("disk", true, (consumerTag, message) -> {
            System.out.println("ReceiveLogsDirect02控制台打印接收到的消息：" + new String(message.getBody()));
        }, consumerTag -> {});
    }
}

‍

生产者

发送多个消息，并且 routing key 不同

package com.peterjxl.rabbitmq.demo7;

import com.peterjxl.rabbitmq.util.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class DirectLogs {


    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        channel.basicPublish(EXCHANGE_NAME,"info",null,"info message".getBytes());
        channel.basicPublish(EXCHANGE_NAME,"warning",null,"warning message".getBytes());
        channel.basicPublish(EXCHANGE_NAME,"error",null,"error message".getBytes());
    }
}

‍

测试

先启动两个消费者，然后启动生产者，运行结果正常：

‍

‍

‍

源码

已将源码上传到 Gitee (opens new window) 或 GitHub (opens new window) 上。并且创建了分支 demo6，读者可以通过切换分支来查看本文的示例代码