RabbitMQ

本文集文档发布于灏天文库

2.5 交换机 (Exchanges)

文档摘要

2.5 交换机 (Exchanges) 2.5 交换机 (Exchanges) 详解 在 RabbitMQ 中，交换机 (Exchanges) 扮演着消息路由的关键角色。它们接收来自生产者 (Producers) 的消息，并根据预定义的规则将这些消息路由到一个或多个队列 (Queues)。 简单来说，交换机是 RabbitMQ 消息传递的核心枢纽，负责将消息从生产者发送到正确的消费者。 2.5.1 交换机的类型 RabbitMQ 提供了几种不同类型的交换机，每种交换机使用不同的路由算法。 选择哪种类型的交换机取决于你的应用场景和路由需求。

2.5 交换机 (Exchanges)

2.5 交换机 (Exchanges) 详解

在 RabbitMQ 中，交换机 (Exchanges) 扮演着消息路由的关键角色。它们接收来自生产者 (Producers) 的消息，并根据预定义的规则将这些消息路由到一个或多个队列 (Queues)。 简单来说，交换机是 RabbitMQ 消息传递的核心枢纽，负责将消息从生产者发送到正确的消费者。

2.5.1 交换机的类型

RabbitMQ 提供了几种不同类型的交换机，每种交换机使用不同的路由算法。 选择哪种类型的交换机取决于你的应用场景和路由需求。 以下是几种常见的交换机类型：

  • Direct Exchange (直连交换机)： Direct Exchange 使用消息的路由键 (Routing Key) 进行精确匹配。 只有当消息的路由键与队列的绑定键 (Binding Key) 完全匹配时，消息才会被路由到该队列。

  • Fanout Exchange (扇形交换机)： Fanout Exchange 将消息广播到所有绑定到它的队列，忽略消息的路由键。 这对于需要将消息同时发送给多个消费者的场景非常有用。

  • Topic Exchange (主题交换机)： Topic Exchange 使用通配符匹配路由键和绑定键。 这提供了更灵活的路由方式，允许根据模式匹配将消息路由到队列。

  • Headers Exchange (头部交换机)： Headers Exchange 根据消息的头部信息进行路由，而不是路由键。 这允许基于消息属性进行更复杂的路由。

2.5.2 交换机的工作原理

  1. 生产者发布消息： 生产者将消息发送到交换机，并指定一个路由键 (Routing Key)。

  2. 交换机接收消息： 交换机接收到消息和路由键。

  3. 交换机根据类型和绑定规则路由消息： 交换机根据自身类型 (Direct, Fanout, Topic, Headers) 和绑定到它的队列的绑定键 (Binding Key) 来决定将消息路由到哪些队列。

  4. 消息被路由到队列： 消息被复制并发送到所有匹配的队列。

  5. 消费者消费消息： 消费者从队列中接收并处理消息。

可以用 Mermaid 图示来表示这个流程:

2.5.3 交换机的声明

在 RabbitMQ 中，交换机需要在使用前进行声明。 声明交换机包括指定交换机的名称、类型和一些可选参数，例如是否持久化。

以下是使用 Java 客户端声明交换机的示例代码：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ExchangeDeclareExample {
    private static final String EXCHANGE_NAME = "my_direct_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // RabbitMQ server host
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明一个 direct 类型的交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true); // durable = true，表示持久化
            System.out.println("Exchange '" + EXCHANGE_NAME + "' declared successfully.");
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}
​

代码解释：

  • ConnectionFactory 用于创建与 RabbitMQ 服务器的连接。

  • channel.exchangeDeclare(EXCHANGE_NAME, "direct", true); 声明了一个名为 my_direct_exchange 的 Direct Exchange。

    • EXCHANGE_NAME: 交换机的名称。

    • "direct": 交换机的类型。

    • true: 表示交换机是持久化的。 持久化的交换机在 RabbitMQ 服务器重启后仍然存在。 如果设置为 false，则交换机在服务器重启后会被删除。

2.5.4 交换机的绑定

交换机需要与队列进行绑定，才能将消息路由到相应的队列。 绑定指定了交换机和队列之间的关系，以及路由键 (Binding Key)。

以下是使用 Java 客户端进行交换机绑定的示例代码：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ExchangeBindExample {
    private static final String EXCHANGE_NAME = "my_direct_exchange";
    private static final String QUEUE_NAME = "my_queue";
    private static final String ROUTING_KEY = "my.routing.key";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明一个队列 (如果队列不存在)
            channel.queueDeclare(QUEUE_NAME, true, false, false, null); // durable = true
            // 将交换机和队列绑定，并指定路由键
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            System.out.println("Queue '" + QUEUE_NAME + "' bound to exchange '" + EXCHANGE_NAME + "' with routing key '" + ROUTING_KEY + "'.");
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}
​

代码解释：

  • channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY); 将名为 my_queue 的队列绑定到名为 my_direct_exchange 的交换机，并指定路由键为 my.routing.key。

    • QUEUE_NAME: 队列的名称。

    • EXCHANGE_NAME: 交换机的名称。

    • ROUTING_KEY: 路由键。 对于 Direct Exchange，只有当消息的路由键与此绑定键完全匹配时，消息才会被路由到该队列。

2.5.5 不同类型交换机的代码实践

以下分别给出 Direct, Fanout, Topic Exchange 的简单示例代码。

2.5.5.1 Direct Exchange 示例

生产者：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class DirectExchangeProducer {
    private static final String EXCHANGE_NAME = "direct_exchange_example";
    private static final String ROUTING_KEY = "info"; // 可以尝试 "warning" 或 "error"
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            String message = "This is an info message.";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + ROUTING_KEY + "':'" + message + "'");
        }
    }
}
​

消费者：

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class DirectExchangeConsumer {
    private static final String EXCHANGE_NAME = "direct_exchange_example";
    private static final String ROUTING_KEY = "info"; // 必须与生产者一致才能收到消息
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
    }
}
​

2.5.5.2 Fanout Exchange 示例

生产者：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class FanoutExchangeProducer {
    private static final String EXCHANGE_NAME = "fanout_exchange_example";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);
            String message = "Hello, everyone!";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes(StandardCharsets.UTF_8)); // 路由键为空
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}
​

消费者：

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class FanoutExchangeConsumer {
    private static final String EXCHANGE_NAME = "fanout_exchange_example";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, ""); // 路由键为空
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
    }
}
​

2.5.5.3 Topic Exchange 示例

生产者：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class TopicExchangeProducer {
    private static final String EXCHANGE_NAME = "topic_exchange_example";
    private static final String ROUTING_KEY = "quick.orange.rabbit"; // 可以尝试 "lazy.orange.elephant" 或 "quick.orange.fox"
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
            String message = "Hello, Topic Exchange!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + ROUTING_KEY + "':'" + message + "'");
        }
    }
}
​

消费者：

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class TopicExchangeConsumer {
    private static final String EXCHANGE_NAME = "topic_exchange_example";
    private static final String BINDING_KEY = "*.orange.*"; // 可以尝试 "#.rabbit" 或 "lazy.#"
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, BINDING_KEY);
        System.out.println(" [*] Waiting for messages with binding key: " + BINDING_KEY + ". To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
    }
}
​

关于 Topic Exchange 的通配符：

  • * (星号): 代表一个单词。

  • # (井号): 代表零个或多个单词。

2.5.6 交换机属性

  • Name (名称): 交换机的唯一标识符。

  • Type (类型): 交换机的类型 (Direct, Fanout, Topic, Headers)。

  • Durable (持久性): 如果设置为 true，则交换机在 RabbitMQ 服务器重启后仍然存在。 默认值为 false。

  • Auto-delete (自动删除): 如果设置为 true，则当最后一个绑定到它的队列取消绑定时，交换机将被自动删除。 默认值为 false。

  • Internal (内部): 如果设置为 true，则客户端无法直接将消息发布到此交换机。 它通常用于交换机之间的路由。

  • Arguments (参数): 用于配置交换机的其他参数，例如备用交换机 (Alternate Exchange)。

2.5.7 总结

交换机是 RabbitMQ 中消息路由的核心组件。 理解不同类型的交换机及其工作原理对于构建高效可靠的消息传递系统至关重要。 通过合理选择交换机类型和配置绑定规则，可以实现灵活的消息路由，满足各种应用场景的需求。 在实际应用中，应该根据具体的业务需求选择合适的交换机类型，并仔细设计路由键和绑定键，以确保消息能够正确地路由到目标队列。

发布者: 作者: 转发
评论区 (0)
U
0/500