RabbitMQ交换机(Exchanges)

news/2024/10/30 22:21:10/

目录

一、概念

二、临时队列

三、绑定

四、Fanout(扇出交换机)

(一)介绍

(二)实战 

五、Direct(直接交换机)

(一)介绍

(二)实战

六、Topic(主题交换机)

(一)介绍

(二)案例

(三)实战

七、Header(头部交换机)


一、概念

        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交换机不常用,这里简单介绍

 


http://www.ppmy.cn/news/28225.html

相关文章

Gradle7.4安装与基本使用

文章目录一.前言二.下载Gradle三.Gradle镜像源-全局级配置四.配置Gradle wrapper-项目级配置五.Gradle对测试的支持五.生命周期5.1 settings文件六.Gradle任务入门6.1 任务行为6.2 任务依赖方式七. Dependencies依赖引入7.1 依赖冲突及解决方案八.Gradle整合多模块SpringBoot九…

【异常检测三件套】系列3--时序异常检测综述

写在前面: 异常检测共包含3个内容,从多个方面剖析异常检测方法,本文为第三篇。过往内容请查看以下链接: 【异常检测三件套】系列1--14种异常检测算法https://blog.csdn.net/allein_STR/article/details/128114175?csdn_share_tail=%7B%22type%22%3A%22blog%22%2C%22rType%…

Linux 进程:理解进程和pcb

目录一、进程的概念二、CPU分时机制三、并发与并行1.并发2.并行四、pcb的概念一、进程的概念 什么是进程&#xff1f; 进程就是进行中的程序&#xff0c;即运行中的应用程序。比如&#xff1a;电脑上打开的LOL、QQ…… 这些都是一个个的进程。 什么是应用程序&#xff1f; 应用…

Scratch少儿编程案例-算法练习-实现华氏和摄氏度温度转换

专栏分享 点击跳转=>Unity3D特效百例点击跳转=>案例项目实战源码点击跳转=>游戏脚本-辅助自动化点击跳转=>Android控件全解手册点击跳转=>Scratch编程案例👉关于作者

JUC包:CountDownLatch源码+实例讲解

1 缘起 有一次听到同事谈及AQS时&#xff0c;我有很多点懵&#xff0c; 只知道入队和出队&#xff0c;CLH&#xff08;Craig&#xff0c;Landin and Hagersten&#xff09;锁&#xff0c;并不了解AQS的应用&#xff0c; 同时结合之前遇到的多线程等待应用场景&#xff0c;发现…

JavaScript 库

JavaScript库是一组封装好的JavaScript函数和对象&#xff0c;可以重复利用和调用&#xff0c;用于简化开发过程中的编码和实现&#xff0c;提高开发效率和可维护性。在实际工作中&#xff0c;JavaScript库是一个非常常用的工具&#xff0c;常用的库包括jQuery、React、Vue.js等…

Python小白入门- 01( 第一章,第1节) 介绍 Python 编程语言

1. 介绍 Python 编程语言 1.1 Python 是什么 Python 是一种高级的、解释型、面向对象的编程语言,具有简洁、易读、易写的语法特点。Python 由 Guido van Rossum 于 1989 年在荷兰创造,并于 1991 年正式发布。 Python 语言广泛应用于数据科学、Web 开发、人工智能、自动化测…

2018年蓝桥杯省赛试题-5道(Python)

文章目录一、日志统计思考二、递增三元组思考三、螺旋折线思考四、乘积最大思考五、全球变暖思考尾声提示&#xff1a;以下是本篇文章正文内容&#xff0c;下面案例可供参考 一、日志统计 题目描述 小明维护着一个程序员论坛。 现在他收集了一份"点赞"日志&#xf…