0
0

RabbitMQ交换机（Exchanges）

一、概念

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

    相反，**生产者只能将消息发送到交换机****(exchange)**，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。**交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。**

交换机的类型：直接(direct), 主题(topic) ,标题(headers) , 扇出(fanout)

**无名 exchange **

第一个参数是交换机的名称。空字符串表示默认或无名称交换机：消息能路由发送到队列中其实

是由 routingKey(bindingkey)绑定 key 指定的，如果它存在的话

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes() );

二、临时队列

    每当我们连接到 Rabbit 时，我们都需要一个全新的空队列，为此我们可以创建一个具有**随机名称****的队列**，或者能让服务器为我们选择一个随机队列名称那就更好了。其次**一旦我们断开了消费者的连****接，队列将被自动删除。 **

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

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

三、绑定

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

四、Fanout（扇出交换机）

（一）介绍

Fanout 这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的

所有队列中。系统中默认有些 exchange 类型

（二）实战

我们声明一个logs交换机以及创建两个临时队列绑定在交换机上，routeKey为空字符串，当EmitLog生产者向logs交换机发消息时两个队列都接收得到。这个即为扇出交换机。

ReceiveLogs01 将接收到的消息打印在控制台 

public class ReceiveLogs01 {

    private final static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        /*
         * 生成一个临时的队列，队列名字随机
         */
        String queueName = channel.queueDeclare().getQueue();
        // 将临时队列绑定在我们定义的交换机上，路由Key为空字符串
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        // 接收消息回调函数
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("ReceiveLogs01控制台打印接收到的消息：" + new String(message.getBody()));
        };
        
        channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
    }
}

ReceiveLogs02 将接收到的消息打印在控制台 

public class ReceiveLogs02 {

    private final static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        /*
         * 生成一个临时的队列，队列名字随机
         */
        String queueName = channel.queueDeclare().getQueue();
        // 将临时队列绑定在我们定义的交换机上，路由Key为空字符串
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("ReceiveLogs02控制台打印接收到的消息：" + new String(message.getBody()));
        };

        channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
    }
}

EmitLog 发送消息给两个消费者接收

public class EmitLog {

    private final static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws IOException {
        // 获取信道
        Channel channel = RabbitMqUtils.getChannel();
        // 交换机声明，在生产者声明消费者方就不用声明了
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 控制台输入内容充当消息发送
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNext()) {
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println("生产者发出消息");
        }
    }
}

总结：

当交换机内的队列routeKey为空字符串时，此时向该交换机发送消息且routeKey也为空字符串的话那么所有的队列都接收得到消息，即为扇出交换机。

五、Direct（直接交换机）

（一）介绍

当我们想对某一个队列单独发送消息，可以根据设置不同的routeKey来达到这种效果，这就是直接交换机。

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

在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列

Q1。绑定键为 black和green 的消息会被发布到队列 Q2，其他消息类型的消息将被丢弃。

多重绑定无非就是绑定的时候多出一条queueBind

channel.queueBind(queue_name, EXCHANGE_NAME, "info");
channel.queueBind(queue_name, EXCHANGE_NAME, "waring");

（二）实战

此时队列的声明需要放在消费者方，因为每个消费者代表不同的队列，对于交换机的声明可以放在生产者

消费者01

public class ReceiveLogsDirect01 {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        String queue_name = "console";
        channel.queueDeclare(queue_name, false, false, false, null);
        channel.queueBind(queue_name, EXCHANGE_NAME, "info");
        channel.queueBind(queue_name, EXCHANGE_NAME, "waring");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("ReceiveLogsDirect01收到消息" + new String(message.getBody()));
        };
        channel.basicConsume(queue_name, deliverCallback, (consumerTag -> {}));
    }
}

消费者02

public class ReceiveLogsDirect02 {

    private static final String EXCHANGE_NAME = "direct_logs";

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

        String queue_name = "disk";
        channel.queueDeclare(queue_name, false, false, false, null);
        channel.queueBind(queue_name, EXCHANGE_NAME, "error");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("ReceiveLogsDirect02收到消息" + new String(message.getBody()));
        };
        channel.basicConsume(queue_name, deliverCallback, (consumerTag -> {}));
    }
}

生产者

public class EmitLogDirect {

    private final static String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        /*
         * 生产者并不知道消息会发向哪个队列，他只是指明找哪个交换机中的路由key
         * 由交换机去寻找队列传递消息
         */

        Scanner sc = new Scanner(System.in);
        while(sc.hasNext()) {
            String message = sc.next();
            channel.basicPublish(EXCHANGE_NAME, "info", null,  message.getBytes());
        }
    }
}

总结：

我们可以通过指定routeKey来向对应的队列单独发消息，即为直接交换机

六、Topic（主题交换机）

（一）介绍

    发送到类型是 topic 交换机的消息的 routing_key 不能随意写，必须满足一定的要求，它**必须是一个单****词列表，以点号分隔开**。这些单词可以是任意单词，比如说："stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit".这种类型的。当然这个单词列表最多不能超过 255 个字节。

在这个规则列表中，其中有两个替换符是大家需要注意的

  • ***(星号)****可以代替一个单词 **
  • **#(井号)****可以替代零个或多个单词 **

（二）案例

队列Q1的routeKey为“*.orange.”，Q2的routeKey有两个，分别为“.*.rabbit”和“lazy.#”

上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的

quick.orange.rabbit 被队列 Q1Q2 接收到

lazy.orange.elephant 被队列 Q1Q2 接收到

quick.orange.fox 被队列 Q1 接收到

lazy.brown.fox 被队列 Q2 接收到

lazy.pink.rabbit 虽然满足两个绑定但只被队列 Q2 接收一次

quick.brown.fox 不匹配任何绑定不会被任何队列接收到会被丢弃

quick.orange.male.rabbit 是四个单词不匹配任何绑定会被丢弃

lazy.orange.male.rabbit 是四个单词但匹配 Q2

当队列绑定关系是下列这种情况时需要引起注意

  • 当一个队列绑定键是**#,****那么这个队列将接收所有数据，就有点像 ****fanout ****了 **
  • 如果队列绑定键当中没有#和***出现，那么该队列绑定类型就是 ****direct ****了 **

（三）实战

消费者01

public class ReceiveLogsTopic01 {

    public static final String EXCHANGE_NAME = "topic_logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = "Q1";
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, EXCHANGE_NAME,  "*.orange.*");

        System.out.println("ReceiveLogsTopic01等待接收消息...");
        DeliverCallback deliverCallback = (consumeTag, message) -> {
            System.out.println("接收队列：" + queueName + "绑定键：" + message.getEnvelope().getRoutingKey());
            System.out.println("接收到消息：" + new String(message.getBody()));
        };

        channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));

    }
}

消费者02

public class ReceiveLogsTopic02 {

    public static final String EXCHANGE_NAME = "topic_logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = "Q2";
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
        channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
        System.out.println("ReceiveLogsTopic02等待接收消息...");
        DeliverCallback deliverCallback = (consumeTag, message) -> {
            System.out.println("接收队列：" + queueName + "绑定键：" + message.getEnvelope().getRoutingKey());
            System.out.println("接收到消息：" + new String(message.getBody()));
        };

        channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
    }
}

生产者

public class EmitLogTopic {

    public static final String EXCHANGE_NAME = "topic_logs";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        Map<String, String> bindingKeyMap = new HashMap<>();
        bindingKeyMap.put("quick.orange.rabbit","被队列 Q1Q2 接收到");
        bindingKeyMap.put("lazy.orange.elephant","被队列 Q1Q2 接收到");
        bindingKeyMap.put("quick.orange.fox","被队列 Q1 接收到");
        bindingKeyMap.put("lazy.brown.fox","被队列 Q2 接收到");
        bindingKeyMap.put("lazy.pink.rabbit","虽然满足两个绑定但只被队列 Q2 接收一次");
        bindingKeyMap.put("quick.brown.fox","不匹配任何绑定不会被任何队列接收到会被丢弃");
        bindingKeyMap.put("quick.orange.male.rabbit","是四个单词不匹配任何绑定会被丢弃");
        bindingKeyMap.put("lazy.orange.male.rabbit","是四个单词但匹配 Q2");

        for(Map.Entry<String, String> bindingKeyMapEntry : bindingKeyMap.entrySet()) {
            String bindingKey = bindingKeyMapEntry.getKey();
            String message = bindingKeyMapEntry.getValue();
            channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes());
            System.out.println("生产者发出消息：" + message);
        }
    }
}

结果

七、Header（头部交换机）

由于header交换机不常用，这里简单介绍

标签： rabbitmq 分布式 java-rabbitmq
本文转载自: https://blog.csdn.net/m0_62946761/article/details/129223083
版权归原作者 zoeil 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ交换机（Exchanges）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

249基于java ssm springboot安全生产培训管理系统学习课件（源码+文档+运行视频+讲解视频）

【Linux庖丁解牛】—软件安装&vim！

【花雕学编程】Arduino LVGL 之在不同平台创建相同的用户界面

基于电商大数据的商品推荐系统

UE5.4像素流及前端接入

Unity中接入讯飞语音（实时语音转写）WebAPI（非SDK）

springboot实现获取客户端IP地址

文章导航