【RabbitMQ五】——RabbitMQ路由模式(Routing)
RabbitMQ路由模式
- 前言
- RabbitMQ模式的基本概念
- 为什么要使用Rabbitmq 路由模式
- RabbitMQ路由模式组成元素
- 路由模式完整代码
- Pom文件引入RabbtiMQ依赖
- RabbitMQ工具类
- 生产者
- 消费者1
- 消费者2
- 运行结果截图
前言
通过本篇博客能够简单使用RabbitMQ的路由模式。
本篇博客主要是博主通过官网以及学习他人的博客总结出的RabbitMQ发布订阅模式。其中如果有误欢迎大家及时指正。
RabbitMQ模式的基本概念
路由模式是根据Routing Key有条件的将消息筛选后发送给消费者,消费者只接受筛选之后的消息
路由模式的核心是:
配置一个类型为direct的交换机,并且需要指定不同的路由键(routing key),把对应的消息从交换机路由到不同的消息队列进行存储,再由对应的消费者进行消费
为什么要使用Rabbitmq 路由模式
由于发布订阅模式是无条件将所有消息分发给所有消费者,路由模式可以根据条件(Routing Key)将消息筛选之后发送给消费者。
应用场景:
例如:有一个股票分析机构,每天都会有一些独家的股票分析报告。对于其他一些应用平台,想要每天都到这家股票分析机构提供的百度的独家股票分析报告,对于另外一些应用平台想要收到谷歌的独家股票分析报告,就可以使用路由模式。
RabbitMQ路由模式组成元素

P:生产者,向交换机发送消息的是否需要指定routing key
X:交换机,接收生产者发送的消息,需要指定交换机的类型为direct,并且将消息发送给与routing key匹配的队列
C1:消费者1,它所在队列指定了需要routing key为error的信息
C2:消费者2,其所在队列指定了需要routing key 为 info、error、warning 的消息
路由模式完整代码
**业务场景:**生产者为日志分发平台,分发info、warning、error级别的日志,消费者1只接受日志级别为error的日志,消费者2接收全部日志。
Pom文件引入RabbtiMQ依赖
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.10.0</version></dependency>
RabbitMQ工具类
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author : [WangWei]* @version : [v1.0]* @className : RabbitMQUtils* @description : [rabbitmq工具类]* @createTime : [2023/1/17 8:49]* @updateUser : [WangWei]* @updateTime : [2023/1/17 8:49]* @updateRemark : [描述说明本次修改内容]*/
public class RabbitMQUtils {/** @version V1.0* Title: getConnection* @author Wangwei* @description 创建rabbitmq连接* @createTime 2023/1/17 8:52* @param []* @return com.rabbitmq.client.Connection*/public static Connection getConnection() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("ip");factory.setPort(5672);factory.setVirtualHost("虚拟主机");factory.setUsername("用户名");factory.setPassword("密码");//创建连接Connection connection=factory.newConnection();return connection;}/** @version V1.0* Title: getChannel* @author Wangwei* @description 创建信道* @createTime 2023/1/17 8:55* @param []* @return com.rabbitmq.client.Channel*/public static Channel getChannel() throws IOException, TimeoutException {Connection connection=getConnection();Channel channel=connection.createChannel();return channel;}
}
生产者
import com.rabbitmq.client.Channel;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;/*** @author : [WangWei]* @version : [v1.0]* @className : Producer* @description : [生产者]* @createTime : [2023/2/1 9:38]* @updateUser : [WangWei]* @updateTime : [2023/2/1 9:38]* @updateRemark : [描述说明本次修改内容]*/
public class Producer {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws IOException, TimeoutException {//建立连接RabbitMQUtils.getConnection();//声明通道Channel channel = RabbitMQUtils.getChannel();//创建fanout类型交换机并命名为logschannel.exchangeDeclare(EXCHANGE_NAME,"direct");//声明routingKeyString severityInfo="info";String severityError="error";String severityWarning="warning";//循环发送2条消息for (int i = 0; i <2 ; i++) {String msg="路由模式info:"+i;/*推送消息*交换机命名,不填写使用默认的交换机* routingKey -路由键-* props:消息的其他属性-路由头等正文* msg消息正文*/channel.basicPublish(EXCHANGE_NAME,severityInfo,null,msg.getBytes(StandardCharsets.UTF_8));System.out.println(msg);}//循环发送2条消息for (int i = 0; i <2 ; i++) {String msg="路由模式error:"+i;/*推送消息*交换机命名,不填写使用默认的交换机* routingKey -路由键-* props:消息的其他属性-路由头等正文* msg消息正文*/channel.basicPublish(EXCHANGE_NAME,severityError,null,msg.getBytes(StandardCharsets.UTF_8));System.out.println(msg);}//循环发送2条消息for (int i = 0; i <2 ; i++) {String msg="路由模式warning:"+i;/*推送消息*交换机命名,不填写使用默认的交换机* routingKey -路由键-* props:消息的其他属性-路由头等正文* msg消息正文*/channel.basicPublish(EXCHANGE_NAME,severityWarning,null,msg.getBytes(StandardCharsets.UTF_8));System.out.println(msg);}}
}
消费者1
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author : [WangWei]* @version : [v1.0]* @className : ConsumerOne* @description : [消费者1]* @createTime : [2023/2/1 9:39]* @updateUser : [WangWei]* @updateTime : [2023/2/1 9:39]* @updateRemark : [描述说明本次修改内容]*/
public class ConsumerOne {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws IOException, TimeoutException {RabbitMQUtils.getConnection();Channel channel = RabbitMQUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME,"direct");String queueName = channel.queueDeclare().getQueue();//声明routingKey (error)String severityError="error";//交换机与队列进行绑定-如果没有队列与交换机进行绑定,那么消费者接受不到生产者的消息,消息会丢失//queueName绑定了direct_logs交换机并且绑定了routingKeychannel.queueBind(queueName, EXCHANGE_NAME,severityError );//因为Rabbitmq服务器将异步地向我们推送消息,所以我们以对象的形式提供了一个回调,该回调将缓冲消息,直到我们准备好使用它们。DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}
消费者2
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author : [WangWei]* @version : [v1.0]* @className : ConsumerTwo* @description : [消费者2]* @createTime : [2023/2/1 9:38]* @updateUser : [WangWei]* @updateTime : [2023/2/1 9:38]* @updateRemark : [描述说明本次修改内容]*/
public class ConsumerTwo {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws IOException, TimeoutException {RabbitMQUtils.getConnection();Channel channel = RabbitMQUtils.getChannel();//创建fanout类型交换机并命名为logschannel.exchangeDeclare(EXCHANGE_NAME,"direct");//创建了一个非持久的、排他的、自动删除的队列,并生成了一个名称String queueName = channel.queueDeclare().getQueue();//声明routingKey (info,error,warning)String severityInfo="info";String severityError="error";String severityWarning="warning";//交换机与队列进行绑定-如果没有队列与交换机进行绑定,那么消费者接受不到生产者的消息,消息会丢失//queueName绑定了direct_logs交换机并且绑定了3个routingKeychannel.queueBind(queueName, EXCHANGE_NAME,severityInfo );channel.queueBind(queueName, EXCHANGE_NAME,severityError );channel.queueBind(queueName, EXCHANGE_NAME,severityWarning );//因为Rabbitmq服务器将异步地向我们推送消息,所以我们以对象的形式提供了一个回调,该回调将缓冲消息,直到我们准备好使用它们。DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}}
运行结果截图



相关文章:
【RabbitMQ五】——RabbitMQ路由模式(Routing)
RabbitMQ路由模式前言RabbitMQ模式的基本概念为什么要使用Rabbitmq 路由模式RabbitMQ路由模式组成元素路由模式完整代码Pom文件引入RabbtiMQ依赖RabbitMQ工具类生产者消费者1消费者2运行结果截图前言 通过本篇博客能够简单使用RabbitMQ的路由模式。 本篇博客主要是博主通过官网…...
【C语言】宏定义 结构体 枚举变量的用法
目录 一、数据类型 二、C语言宏定义 三、C语言typedef重命名 四、 #define与typedef的区别 五、结构体 六、枚举变量 补充学习一点STM32的必备基础知识 一、数据类型 二、C语言宏定义 关键字:#define 用途:用一个字符串代替一个数字,…...
锁升级之Synchronized
Synchronized JVM系统锁一个对象里如果有多个synchronized方法,同一时刻,只要有一个线程去调用其中的一个synchronized方法,其他线程只能等待!锁的是当前对象,对象被锁定后,其他线程都不能访问当前对象的其…...
基于nodejs+vue疫情网课管理系统
疫情网课也都将通过计算机进行整体智能化操作,对于疫情网课管理系统所牵扯的管理及数据保存都是非常多的,例如管理员:首页、个人中心、学生管理、教师管理、班级管理、课程分类管理、课程表管理、课程信息管理、作业信息管理、请假信息管理、上课签到管理、论坛交流…...
Zabbix 构建监控告警平台(三)
Zabbix User parametersZabbix Trigger1.Zabbix User parameters 1.1即自定义KEY 注意:mysql安装在被监测主机 [rootlocalhost ~]# yum -y install mariadb-server mariadb [rootlocalhost ~]# systemctl start mariadb [rootlocalhost ~]# mysqladmin -uroot statu…...
Linux系统之dool命令行工具的基本使用
Linux系统之dool命令行工具的基本使用一、dool命令行工具介绍二、本地系统环境检查1.检查系统版本2.检查系统内核版本三、下载dool软件包1.创建下载目录2.下载dool四、安装dool1.安装python32.安装dool五、dool的命令帮助六、dool的基本使用1.直接使用dool监控系统2.监控cpu和网…...
LeetCode-2335-装满杯子需要的最短总时长
1、堆 我们可以维护一个堆,首先我们将数组中不为0的数全部加入堆中,而后进行循环。当堆不为空时,我们将堆顶元素出堆并减一,而后观察是否还能继续出堆,若能则出堆,否则跳过,最后我们将处理后的…...
npm ERR! code ELIFECYCLE解决方案,npm犯错!myweb@1.0.0构建脚本失败。
1.问题npm ERR! code ELIFECYCLEnpm ERR! errno 1npm ERR! myweb1.0.0 build: webpack --config config/webpack.config.jsnpm ERR! Exit status 1npm ERR!npm ERR! Failed at the myweb1.0.0 build script.npm犯错!代码ELIFECYCLEnpm犯错!errno 1npm犯错!myweb1.0.0 build: we…...
最小二乘支持向量机”在学习偏微分方程 (PDE) 解方面的应用(Matlab代码实现)
目录 💥1 概述 📚2 运行结果 🎉3 参考文献 👨💻4 Matlab代码 💥1 概述 本代码说明了“最小二乘支持向量机”在学习偏微分方程 (PDE) 解方面的应用。提供了一个示例,…...
ISYSTEM调试实践8-winIDEA Analyzer功能1
前面几篇介绍了ISYSTEM的基本调试界面和功能,相比我之前用过的IDE,除了几种断点方式和脚本功能以外,应该都是比较简单,稍微操作一下就可以直接上手,后续我将介绍winIDEA的Analyzer 功能。 1 Analyzer简介 iSYSTEM An…...
每日学术速递2.11
CV - 计算机视觉 | ML - 机器学习 | RL - 强化学习 | NLP 自然语言处理 Subjects: cs.IR、cs.MM 1.A Comprehensive Survey on Multimodal Recommender Systems: Taxonomy, Evaluation, and Future Directions 标题:关于多模态推荐系统的综合调查:分…...
宝塔搭建实战php开源likeadmin通用管理admin端vue3源码(二)
大家好啊,我是测评君,欢迎来到web测评。 上一期给大家分享了server端的部署方式,今天来给大家分享admin端在本地搭建,与打包发布到宝塔的方法。感兴趣的朋友可以自行下载学习。 技术架构 vscode node16 vue3 elementPlus vit…...
网络基础-虚拟化工具-网桥
系列文章目录 本系列文章主要是回顾和学习工作中常用的网络基础命令,在此记录以便于回顾。 该篇文章主要是讲解虚拟化的工具网桥相关的概念和常用命令 提示:写完文章后,目录可以自动生成,如何生成可参考右边的帮助文档 文章目录系…...
剑指 Offer 14- II. 剪绳子 II
剑指 Offer 14- II. 剪绳子 II 给你一根长度为 n 的绳子,请把绳子剪成整数长度的 m 段(m、n都是整数,n>1并且m>1),每段绳子的长度记为 k[0],k[1]…k[m - 1] 。请问 k[0]k[1]…*k[m - 1] 可能的最大乘积是多少&a…...
English Learning - Day55 作业打卡 2023.2.9 周四
English Learning - Day55 作业打卡 2023.2.9 周四引言1. Jim 在看电视的时候他的老婆正在做饭。2. 他刚睡着电话就响了。3. 我正在想事情,这时忽然有人从后面抓我胳膊。4. 我们总是边吃火锅边唱歌。5. 他一听说出了事故,马上就来了现场。6. He entered …...
pixhawk2.4.8-地面站配置-APM固件
文章目录一、硬件准备二、软件准备1 已实飞测试2 MP地面站 任意版本下载:3 APM固件 任意版本下载:三、飞控校准1 刷固件2 机架选择3 加速度计校准4 指南针校准5 遥控器校准6 飞行模式7 紧急断电&无头模式8 基础参数设置9 电流计校准10 电调校准11 起…...
golang 通道类型
文章目录一、什么是通道类型二、通道产生的原因三、声明channel四、创建channel五、channel相关操作1、发送值2、接收值3、关闭通道3.1 注意3.2 特点四、通道类型1、无缓冲通道2、有缓冲通道五、单向通道一、什么是通道类型 Go 语言中的通道(channel)是一…...
并发、并行、吞吐量、延迟、响应时间 含义理解
并发、并行、吞吐量、延迟、响应时间 知识点了解 1. 响应时间(RT) 理解:响应时间是指系统对请求作出响应的时间。例如一个正在运行的服务,服务内程序接受到参数请求开始,到程序计算完,并将结果返回出去结束,这段时间…...
HTTP 和 HTTPS 的区别
文章目录前言一、HTTP 与 HTTPS 的基本概念HTTPHTTPS二、HTTP 和 HTTPS协议的区别前言 浏览网站时,我们会发现网址有两种格式,一种以http://开头,一种https://开头。好像这两种格式差别不大,只多了一个s,实际上他们有…...
微搭低代码从入门到精通07-基础布局组件
低码开发不同于传统开发,传统开发我们通常需要编写前端代码和后端代码。前端代码由HTML、CSS和JavaScript组成,后端代码我们通常要用后端语言比如Java来编写接口。 低码开发的特点是可视化开发,在编辑器中通过组件的拖拽来完成页面的编制。如…...
Java 语言特性(面试系列2)
一、SQL 基础 1. 复杂查询 (1)连接查询(JOIN) 内连接(INNER JOIN):返回两表匹配的记录。 SELECT e.name, d.dept_name FROM employees e INNER JOIN departments d ON e.dept_id d.dept_id; 左…...
Xshell远程连接Kali(默认 | 私钥)Note版
前言:xshell远程连接,私钥连接和常规默认连接 任务一 开启ssh服务 service ssh status //查看ssh服务状态 service ssh start //开启ssh服务 update-rc.d ssh enable //开启自启动ssh服务 任务二 修改配置文件 vi /etc/ssh/ssh_config //第一…...
CMake基础:构建流程详解
目录 1.CMake构建过程的基本流程 2.CMake构建的具体步骤 2.1.创建构建目录 2.2.使用 CMake 生成构建文件 2.3.编译和构建 2.4.清理构建文件 2.5.重新配置和构建 3.跨平台构建示例 4.工具链与交叉编译 5.CMake构建后的项目结构解析 5.1.CMake构建后的目录结构 5.2.构…...
为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?
在建筑行业,项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升,传统的管理模式已经难以满足现代工程的需求。过去,许多企业依赖手工记录、口头沟通和分散的信息管理,导致效率低下、成本失控、风险频发。例如&#…...
Opencv中的addweighted函数
一.addweighted函数作用 addweighted()是OpenCV库中用于图像处理的函数,主要功能是将两个输入图像(尺寸和类型相同)按照指定的权重进行加权叠加(图像融合),并添加一个标量值&#x…...
在 Nginx Stream 层“改写”MQTT ngx_stream_mqtt_filter_module
1、为什么要修改 CONNECT 报文? 多租户隔离:自动为接入设备追加租户前缀,后端按 ClientID 拆分队列。零代码鉴权:将入站用户名替换为 OAuth Access-Token,后端 Broker 统一校验。灰度发布:根据 IP/地理位写…...
Qt Http Server模块功能及架构
Qt Http Server 是 Qt 6.0 中引入的一个新模块,它提供了一个轻量级的 HTTP 服务器实现,主要用于构建基于 HTTP 的应用程序和服务。 功能介绍: 主要功能 HTTP服务器功能: 支持 HTTP/1.1 协议 简单的请求/响应处理模型 支持 GET…...
linux 下常用变更-8
1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行,YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID: YW3…...
均衡后的SNRSINR
本文主要摘自参考文献中的前两篇,相关文献中经常会出现MIMO检测后的SINR不过一直没有找到相关数学推到过程,其中文献[1]中给出了相关原理在此仅做记录。 1. 系统模型 复信道模型 n t n_t nt 根发送天线, n r n_r nr 根接收天线的 MIMO 系…...
laravel8+vue3.0+element-plus搭建方法
创建 laravel8 项目 composer create-project --prefer-dist laravel/laravel laravel8 8.* 安装 laravel/ui composer require laravel/ui 修改 package.json 文件 "devDependencies": {"vue/compiler-sfc": "^3.0.7","axios": …...
