RabbitMQ-交换机

ops/2024/10/18 0:23:18/

文章目录

    • 交换机
      • fanout
      • Direct
      • topic
      • Headers
      • RPC

交换机

**交换机 **是消息队列中的一个组件，其作用类似于网络路由器。它负责将我们发送的消息转发到相应的目标，就像快递站将快递发送到对应的站点，或者网络路由器将网络请求转发到相应的服务器或客户端一样。交换机的主要功能是提供转发消息的能力，根据消息的路由规则将消息投递到合适的队列或绑定的消费者。
我们可以理解为，如果说一个快递站已经承受不了那么多的快递了，就建多个快递站。

fanout

扇出，广播
特点：消息会被转发到所有绑定到该交换机的队列
场景:很适用于发布订阅的场景，比如写日志，可以多个系统间共享

示例场景：
image.png
生产者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class FanoutProducer {// 交换机名字private static final String EXCHANGE_NAME = "fanout-exchange";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 创建交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String message = scanner.nextLine();channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}}
}

消费者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class FanoutConsumer {//交换机名字private static final String EXCHANGE_NAME = "fanout-exchange";public static void main(String[] argv) throws Exception {//建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();//创建频道Channel channel1 = connection.createChannel();// 声明交换机channel1.exchangeDeclare(EXCHANGE_NAME, "fanout");// 创建队列1，连接到交换机上String queueName = "xiaowang_queue";channel1.queueDeclare(queueName, true, false, false, null);channel1.queueBind(queueName, EXCHANGE_NAME, "");// 创建队列2,连接到交换机上String queueName2 = "xiaoli_queue";channel1.queueDeclare(queueName2, true, false, false, null);channel1.queueBind(queueName2, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 创建交付回调函数1DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [小王] Received '" + message + "'");};// 创建交付回调函数2DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [小李] Received '" + message + "'");};// 开始消费消息队列1channel1.basicConsume(queueName, true, deliverCallback1, consumerTag -> { });// 开始消费消息队列2channel1.basicConsume(queueName2, true, deliverCallback2, consumerTag -> { });}
}

Direct

官方教程：https://www.rabbitmq.com/tutorials/tutorial-four-java.html
特点：消息会根据路由键转发到指定的队列
场景：特定的消息只交给特定的系统（程序）来处理

注意：不同队列可以绑定相同的路由键

示例场景：
老板在发送消息同时会带上路由键，根据路由键找对应的队列来发送
image.png
生产者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class DirectProducer {private static final String EXCHANGE_NAME = "direct-exchange";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//声明交换机是directchannel.exchangeDeclare(EXCHANGE_NAME, "direct");//输入消息 和 路由键Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String userInput = scanner.nextLine();String[] strings = userInput.split(" ");if (strings.length < 1) {continue;}String message = strings[0];String routingKey = strings[1];//发布消息的时候注意指定路由键channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + " with routing:" + routingKey + "'");}}}
}

消费者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class DirectConsumer {private static final String EXCHANGE_NAME = "direct-exchange";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, "direct");// 创建队列String queueName = "xiaoyu_queue";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, EXCHANGE_NAME, "xiaoyu"); //指定2交换机和路由键// 创建队列，随机分配一个队列名称String queueName2 = "xiaopi_queue";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, EXCHANGE_NAME, "xiaopi");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback xiaoyuDeliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [xiaoyu] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback xiaopiDeliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [xiaopi] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, xiaoyuDeliverCallback, consumerTag -> {});channel.basicConsume(queueName2, true, xiaopiDeliverCallback, consumerTag -> {});}
}

topic

官方教程：https://www.rabbitmq.com/tutorials/tutorial-five-java.html
特点：消息会根据一个模糊的路由键转发到指定的队列
场景:特定的一类消息可以交给特定的一类系统（程序）来处理
规则：

  1. :匹配一个单词，比如.orange，那么 abc.orange、ikun.orange 都能匹配
  2. #:匹配0个或多个单词，比如orange.#，那么orange，orange.abc.ikun都能匹配


应用场景：
老板要下发一个任务，让多个组来处理
image.png
生产者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class TopicProducer {private static final String EXCHANGE_NAME = "topic-exchange";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "topic");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String userInput = scanner.nextLine();String[] strings = userInput.split(" ");if (strings.length < 1) {continue;}String message = strings[0];String routingKey = strings[1];channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + " with routing:" + routingKey + "'");}}}
}

消费者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class TopicConsumer {private static final String EXCHANGE_NAME = "topic-exchange";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, "topic");// 创建队列String queueName = "frontend_queue";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, EXCHANGE_NAME, "#.前端.#");// 创建队列String queueName2 = "backend_queue";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, EXCHANGE_NAME, "#.后端.#");// 创建队列String queueName3 = "product_queue";channel.queueDeclare(queueName3, true, false, false, null);channel.queueBind(queueName3, EXCHANGE_NAME, "#.产品.#");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback xiaoaDeliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [xiaoa] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback xiaobDeliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [xiaob] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback xiaocDeliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [xiaoc] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, xiaoaDeliverCallback, consumerTag -> {});channel.basicConsume(queueName2, true, xiaobDeliverCallback, consumerTag -> {});channel.basicConsume(queueName3, true, xiaocDeliverCallback, consumerTag -> {});}
}

这样生产者发消息：前端.后端
image.png
就可以匹配到前端和后端两个队列
image.png

Headers

可以根据headers中的内容来指定发送到哪个队列，由于性能差，比较复杂，一般不推荐使用

RPC

支持用消息队列来模拟RPC的调用，但是一般没必要，直接用 Dubbo、GRPC 等 RPC 框架就好了。


http://www.ppmy.cn/ops/6483.html

相关文章

小成本搏大流量：微信/支付宝小程序搜索排名优化

小成本搏大流量：微信/支付宝小程序搜索排名优化

随着移动互联网的快速发展&#xff0c;小程序已成为企业和个人开发者重要的流量入口和业务承载平台。而小程序搜索排名则是影响小程序曝光量、用户获取及业务转化的关键因素。小柚在本文和大家探讨如何制定有效的优化方案&#xff0c;提升小程序在搜索结果中的排名。 首先跟我…
阅读更多...
Axure引用ECharts图表 解决火狐浏览器出错

Axure引用ECharts图表 解决火狐浏览器出错

Axure原型添加Echarts图表&#xff0c;没耐心看文章的可以直接下载示例 Axure中使用ECharts图表示例 1. 打开Axure新建页面 2. 页面添加元件 元件类型随意&#xff0c;矩形、动态面板、热区、图片 甚至段落都可以3. 命名元件 随意命名&#xff0c;单个页面用到多个图表时名…
阅读更多...
机器学习理论基础—支持向量机的推导（一）

机器学习理论基础—支持向量机的推导（一）

机器学习理论基础—支持向量机的推导 算法原理 SVM:从几何角度&#xff0c;对于线性可分数据集&#xff0c;支持向量机就是找距离正负样本都最远的超平面&#xff0c;相比于感知机&#xff0c;其解是唯一的&#xff0c;且不偏不倚&#xff0c;泛化性能更好。 超平面 n维空间…
阅读更多...
刷题DAY58 | LeetCode 739-每日温度 496-下一个更大元素 I

刷题DAY58 | LeetCode 739-每日温度 496-下一个更大元素 I

739 每日温度&#xff08;medium&#xff09; 给定一个整数数组 temperatures &#xff0c;表示每天的温度&#xff0c;返回一个数组 answer &#xff0c;其中 answer[i] 是指对于第 i 天&#xff0c;下一个更高温度出现在几天后。如果气温在这之后都不会升高&#xff0c;请在…
阅读更多...
集成学习-Bagging与随机森林回归

集成学习-Bagging与随机森林回归

reg_fRFR() reg_tDTR()#实例化决策树 cvKFold(n_splits5,shuffleTrue,random_state1412)#实例化验证方式 result_tcross_validate(reg_t#要进行交叉验证的评估器,X,y,cvcv,scoringneg_mean_squared_error#评估指标,return_train_scoreTrue#是否返回训练分数&#xff0c;后面这几…
阅读更多...
GIS 数据格式转换

GIS 数据格式转换

1、在线工具 mapshaper 2、数据上传 3、数据格式转换 导入数据可导出为多种格式&#xff1a;Shapefile、Json、GeoJson、CSV、TopJSON、KML、SVG
阅读更多...
数据库基本概念和SQL基本语句

数据库基本概念和SQL基本语句

数据库&#xff08;Database&#xff09;是按照数据结构来组织、存储和管理数据的仓库。在数据库中&#xff0c;数据通常以表格的形式存储&#xff0c;这些表格包含了行和列。行通常代表记录&#xff0c;而列代表记录中的不同字段。数据库的设计允许对数据进行高效地查询、更新…
阅读更多...
prompt提示词：影响力营销文案，让AI 帮你写营销文案

prompt提示词：影响力营销文案，让AI 帮你写营销文案

影响力营销文案提问技巧 1&#xff0e;我正在寻找一个有影响力的营销活动大纲&#xff0c;向我的[理想客户角色]展示我的[产品/服务]&#xff0c;并说服他们在符合我们品牌价值的[有影响力的类型]的帮助下采取[期望的行动] 2&#xff0e;我需要一个有影响力的营销活动大纲&am…
阅读更多...
最新文章

Copyright @ 2022~2023