当前位置: 首页 > news >正文

消息队列——rabbitmq的不同工作模式

目录

Work queues 工作队列模式

 Pub/Sub 订阅模式

Routing路由模式

Topics通配符模式 

 工作模式总结


Work queues 工作队列模式

C1和C2属于竞争关系,一个消息只有一个消费者可以取到。

 代码部分只需要用两个消费者进程监听同一个队里即可。

两个消费者呈现竞争关系。

用一个生产者推送10条消息

        for(int i=0;i<10;i++){String body=i+"hello rabbitmq!!!";channel.basicPublish("","work_queues",null,body.getBytes());}

两个监听的消费者接收情况如下。 

 

 Pub/Sub 订阅模式

一个生产者发送消息后有两个消费者可以收到消息。

生产者把消息发给交换机,交换机再把消息通过Routes路由分发给不同的队列。

//发送消息
public class producer_PubSub {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();/** exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)* 参数:* 1.exchange  : 交换价名称* 2.type      : 交换机类型 ,有四种*               DIRECT("direct"),  定向FANOUT("fanout"),   扇形(广播),发送消息到每一个与之绑定队列TOPIC("topic"),     通配符的方式HEADERS("headers"); 参数匹配*3.durable  :是否持久化* 4.autoDelete:是否自动删除* 5.internal: 内部使用。一般false* 6.arguments:参数* *///5.创建交换机String exchangeName="test_fanout";channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);//6.创建队列String queue1Name="test_fanout_queue1";String queue2Name="test_fanout_queue2";channel.queueDeclare(queue1Name,true,false,false,null);channel.queueDeclare(queue2Name,true,false,false,null);/** queueBind(String queue, String exchange, String routingKey)* 参数:* queue:队列名* exchange:交换机名称* routingKey:路由键,绑定规则*   如果交换机类型为fanout,routingKey设置为""* *///7.绑定队列和交换机channel.queueBind(queue1Name,exchangeName,"");channel.queueBind(queue2Name,exchangeName,"");String body="日志信息:调用了findAll方法";//8.发送消息channel.basicPublish(exchangeName,"",null,body.getBytes());//9.释放资源channel.close();connection.close();}
}

 运行之后两个队列里面就会多一条消息

两个消费者的代码大同小异,只是绑定的队列名不同,这里只给其中一个

public class consumer_PubSub1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();String queue1Name="test_fanout_queue1";String queue2Name="test_fanout_queue2";/** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数:*   1.队列名称*   2.autoAck:是否自动确认*   3.callback:回调对象* *///6.接收消息Consumer consumer=new DefaultConsumer(channel){/** 回调方法,当收到消息后,会自动执行该方法* 1.consumerTag:标识* 2.envelope :获取一些信息,交换机,路由key...* 3.properties: 配置信息* 4.body: 数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("properties:"+properties);System.out.println("body:"+new String(body));System.out.println("将日志信息打印到控制台......");}};channel.basicConsume(queue1Name,true,consumer);//不需要关闭资源}
}

 控制台输出有

Routing路由模式

对于特定级别的信息会发送到别的队列,如上图的error,在发送消息时也会有一个routing,只要和后面的队列对应上就可以发送到对应队列。 

 生产者代码:

//发送消息
public class producer_Routing {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();/** exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)* 参数:* 1.exchange  : 交换价名称* 2.type      : 交换机类型 ,有四种*               DIRECT("direct"),  定向FANOUT("fanout"),   扇形(广播),发送消息到每一个与之绑定队列TOPIC("topic"),     通配符的方式HEADERS("headers"); 参数匹配*3.durable  :是否持久化* 4.autoDelete:是否自动删除* 5.internal: 内部使用。一般false* 6.arguments:参数* *///5.创建交换机String exchangeName="test_direct";channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);//6.创建队列String queue1Name="test_direct_queue1";String queue2Name="test_direct_queue2";channel.queueDeclare(queue1Name,true,false,false,null);channel.queueDeclare(queue2Name,true,false,false,null);/** queueBind(String queue, String exchange, String routingKey)* 参数:* queue:队列名* exchange:交换机名称* routingKey:路由键,绑定规则*   如果交换机类型为fanout,routingKey设置为""* *///7.绑定队列和交换机//队列1绑定errorchannel.queueBind(queue1Name,exchangeName,"error");//队列2绑定error,info,warningchannel.queueBind(queue2Name,exchangeName,"info");channel.queueBind(queue2Name,exchangeName,"error");channel.queueBind(queue2Name,exchangeName,"warning");String body="日志信息:调用了findAll方法,级别:info,error,warning";//8.发送消息channel.basicPublish(exchangeName,"error",null,body.getBytes());//9.释放资源channel.close();connection.close();}
}

消费者代码(两个消费者就绑定队列名不一样):

public class consumer_Routing1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();String queue1Name="test_direct_queue1";String queue2Name="test_direct_queue2";/** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数:*   1.队列名称*   2.autoAck:是否自动确认*   3.callback:回调对象* *///6.接收消息Consumer consumer=new DefaultConsumer(channel){/** 回调方法,当收到消息后,会自动执行该方法* 1.consumerTag:标识* 2.envelope :获取一些信息,交换机,路由key...* 3.properties: 配置信息* 4.body: 数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("properties:"+properties);System.out.println("body:"+new String(body));System.out.println("将日志信息存储到数据库");}};channel.basicConsume(queue1Name,true,consumer);//不需要关闭资源}
}

Topics通配符模式 

发送消息时设定的routingkey会和后面的routingkey进行匹配。

生产者代码:

//发送消息
public class producer_Topic {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();/** exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)* 参数:* 1.exchange  : 交换价名称* 2.type      : 交换机类型 ,有四种*               DIRECT("direct"),  定向FANOUT("fanout"),   扇形(广播),发送消息到每一个与之绑定队列TOPIC("topic"),     通配符的方式HEADERS("headers"); 参数匹配*3.durable  :是否持久化* 4.autoDelete:是否自动删除* 5.internal: 内部使用。一般false* 6.arguments:参数* *///5.创建交换机String exchangeName="test_topic";channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);//6.创建队列String queue1Name="test_topic_queue1";String queue2Name="test_topic_queue2";channel.queueDeclare(queue1Name,true,false,false,null);channel.queueDeclare(queue2Name,true,false,false,null);/** queueBind(String queue, String exchange, String routingKey)* 参数:* queue:队列名* exchange:交换机名称* routingKey:路由键,绑定规则*   如果交换机类型为fanout,routingKey设置为""* *///7.绑定队列和交换机// routing key 系统的名称.日志的级别。//需求:所有error级别的日志存入数据库,所有order系统的日志存入数据库channel.queueBind(queue1Name,exchangeName,"#.error");channel.queueBind(queue1Name,exchangeName,"order.*");channel.queueBind(queue2Name,exchangeName,"*.*");String body="日志信息:调用了findAll方法";//8.发送消息channel.basicPublish(exchangeName,"goods.error",null,body.getBytes());//9.释放资源channel.close();connection.close();}
}

 消费者代码

public class consumer_Topic1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory=new ConnectionFactory();//2.设置参数factory.setHost(""); //设置ip地址。默认为127.0.0.1factory.setPort(5672);              //端口 默认值5672factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/factory.setUsername("yhy");        //用户名,默认值guestfactory.setPassword("");     //密码,默认值guest//3.创建连接ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.createChannel();String queue1Name="test_topic_queue1";String queue2Name="test_topic_queue2";/** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数:*   1.队列名称*   2.autoAck:是否自动确认*   3.callback:回调对象* *///6.接收消息Consumer consumer=new DefaultConsumer(channel){/** 回调方法,当收到消息后,会自动执行该方法* 1.consumerTag:标识* 2.envelope :获取一些信息,交换机,路由key...* 3.properties: 配置信息* 4.body: 数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("properties:"+properties);System.out.println("body:"+new String(body));System.out.println("将日志信息存储到数据库");}};channel.basicConsume(queue1Name,true,consumer);//不需要关闭资源}
}

 工作模式总结

相关文章:

消息队列——rabbitmq的不同工作模式

目录 Work queues 工作队列模式 Pub/Sub 订阅模式 Routing路由模式 Topics通配符模式 工作模式总结 Work queues 工作队列模式 C1和C2属于竞争关系&#xff0c;一个消息只有一个消费者可以取到。 代码部分只需要用两个消费者进程监听同一个队里即可。 两个消费者呈现竞争关…...

QT实现用户登录注册功能

本文实例为大家分享了QT实现用户登录注册的具体代码&#xff0c;供大家参考&#xff0c;具体内容如下 1、login.h ? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 #ifndef LOGIN_H #define LOGIN_H #include <QWidget> namespace Ui { c…...

Docker--harbor私有仓库部署与管理

目录 一、构建私有库 1.下载 registry 镜像 2.在 daemon.json 文件中添加私有镜像仓库地址 3.运行 registry 容器 4.为镜像打标签 5.上传到私有仓库 6.列出私有仓库的所有镜像 7.列出私有仓库的centos镜像有哪些tag 8.测试私有仓库下载 二、Harbor 简介 1.什么是Harb…...

idea复制一份web服务在不同端口启动

Idea 运行多个微服务 Idea 一个服务开启多个端口运行 idea 开启多个端口服务_idea开启多个服务_HaHa_Sir的博客-CSDN博客 IntelliJ IDEA 中一个服务按多个端口同时启动与显示Services面板_一个服务多个端口_Touch&的博客-CSDN博客 Idea中一个服务按多个端口同时启动_idea…...

CRM系统化整合从N-1做减法实践 | 京东物流技术团队

1 背景 京销易系统已经接入大网、KA以及云仓三个条线商机&#xff0c;每个条线商机规则差异比较大&#xff0c;当前现状是独立实现三套系统分别做支撑。 2 目标 2022年下半年CRM目标是完成9个新条线业务接入&#xff0c;完成销售过程线上化&#xff0c;实现销售规则统一。 …...

STM32CUBUMX配置RS485(中断接收)--保姆级教程

———————————————————————————————————— ⏩ 大家好哇&#xff01;我是小光&#xff0c;嵌入式爱好者&#xff0c;一个想要成为系统架构师的大三学生。 ⏩最近在开发一个STM32H723ZGT6的板子&#xff0c;使用STM32CUBEMX做了很多驱动&#x…...

苹果iOS 16.6 RC发布:或为iPhone X/8系列养老版本

今天苹果向iPhone用户推送了iOS 16.6 RC更新(内部版本号&#xff1a;20G75)&#xff0c;这是时隔两个月的首次更新。 按照惯例RC版基本不会有什么问题&#xff0c;会在最近一段时间内直接变成正式版&#xff0c;向所有用户推送。 需要注意的是&#xff0c;鉴于iOS 17正式版即将…...

【100天精通python】Day16:python 模块的搜索目录和导入模块异常时的处理方法

目录 1 搜索模块所在目录 2 模块不在搜索目录中 2.1 添加模块所在的目录到PYTHONPATH环境变量 2.2 修改sys.path 2.3 使用绝对路径导入 2.4将模块复制到Python搜索路径中的任意一个目录 2.5 总结 3 其他导入的模块异常处理 3.1 模块未安装 3.2 模块名称拼写错误 3.3模…...

SOC FPGA介绍及开发设计流程

目录 一、SoC FPGA简介 二、SoC FPGA开发流程 2.1 硬件开发 2.2 软件开发 一、SoC FPGA简介 SOC FPGA是在FPGA架构中集成了基于ARM的硬核处理器系统(HPS)&#xff0c;包括处理器、外设和存储器控制器。相较于传统的仅有ARM处理器或 FPGA 的嵌入式芯片&#xff0c;SOC FPGA既…...

MySQL vs. Oracle: 函数比较与联系

引言 MySQL和Oracle是两个广泛使用的关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;它们提供了丰富的函数库来处理和操作数据。本文将详细介绍MySQL和Oracle的所有函数&#xff0c;并逐一介绍它们的相同和不同之处&#xff0c;以帮助读者更好地理解和使用这两…...

【Django学习】(十五)API接口文档平台_项目流程分析_日志器_认证_授权

一、API接口文档平台 使用API接口文档不经可以很好的的维护接口数据&#xff0c;还给测试人员的接口测试工作带来了便利&#xff1b; 我们可以在全局配置文件中添加路由路径生成接口文档 1、使用docs接口文档维护接口 1.1在全局配置文件里指定用于支持coreapi的Schema # 指…...

经营简报及考核360表格

文章目录 经营简报效果图代码tableObjectSpanMethod.js 考核360委员会效果图 经营简报效果图不需要合并单元格且有汇总表头的 懒得封装了&#xff0c;所以整体没有封装 经营简报 效果图 代码 <template><el-tableref"tableRef":data"tableData.lengt…...

Spring Security 构建基于 JWT 的登录认证

一言以蔽之&#xff0c;JWT 可以携带非敏感信息&#xff0c;并具有不可篡改性。可以通过验证是否被篡改&#xff0c;以及读取信息内容&#xff0c;完成网络认证的三个问题&#xff1a;“你是谁”、“你有哪些权限”、“是不是冒充的”。 为了安全&#xff0c;使用它需要采用 …...

PyTorch从零开始实现Transformer

文章目录 自注意力Transformer块编码器解码器块解码器整个Transformer参考来源全部代码&#xff08;可直接运行&#xff09; 自注意力 计算公式 代码实现 class SelfAttention(nn.Module):def __init__(self, embed_size, heads):super(SelfAttention, self).__init__()self.e…...

运动蓝牙耳机什么牌子的好用、最好用的运动蓝牙耳机推荐

音乐是运动的灵魂&#xff0c;而一款优秀的运动耳机则是让音乐与我们的身体完美融合的关键。今天&#xff0c;我推荐五款备受运动爱好者喜爱的耳机&#xff0c;它们以卓越的音质、舒适的佩戴和出色的稳定性能脱颖而出&#xff0c;助你在运动中创造最佳状态。 1、NANK南卡Runne…...

HTTP、HTTPS协议详解

文章目录 HTTP是什么报文结构请求头部响应头部 工作原理用户点击一个URL链接后&#xff0c;浏览器和web服务器会执行什么http的版本持久连接和非持久连接无状态与有状态Cookie和Sessionhttp方法&#xff1a;get和post的区别 状态码 HTTPS是什么ssl如何搞到证书nginx中的部署 加…...

【算法与数据结构】222、LeetCode完全二叉树的节点个数

文章目录 一、题目二、一般遍历解法三、利用完全二叉树性质四、完整代码 所有的LeetCode题解索引&#xff0c;可以看这篇文章——【算法和数据结构】LeetCode题解。 一、题目 二、一般遍历解法 思路分析&#xff1a;利用层序遍历&#xff0c;然后用num记录节点数量。其他的例如…...

登录和注册表单的11个HTML最佳实践

原文&#xff1a;11 HTML best practices for login & sign-up forms 原作者&#xff1a;Andrey Sitnik 翻译已获原文作者许可&#xff0c;禁止转载和商用 大多数网站都有登录或注册表单;它们是业务转换的关键部分。然而&#xff0c;即使是流行的站点也没有实现本文中提到的…...

Mysql删除历史数据

Mysql定时删除历史数据 实现 1.创建存储过程&#xff08;函数&#xff09; SQL DROP PROCEDURE IF EXISTS KeepDatasWith30Days CREATE PROCEDURE KeepDatasWith30Days() BEGINSELECT maxId:max(Id) FROM tableName WHERE CreateTime<DATE(DATE_SUB(NOW(),INTERVAL 31 D…...

Python—数据结构(一)

先放一张自己学习和整理归纳的思维导图&#xff0c;以便让大家都知道我自己的整体学习路线。 数据结构的学习路上内容枯燥&#xff0c;但坚持下来一定有很大的收获&#xff01;加油&#x1f4aa;&#x1f3fb;&#xff01; 数据结构 数据的概念数据元素&#xff1a; 若干基本…...

内存分配函数malloc kmalloc vmalloc

内存分配函数malloc kmalloc vmalloc malloc实现步骤: 1)请求大小调整:首先,malloc 需要调整用户请求的大小,以适应内部数据结构(例如,可能需要存储额外的元数据)。通常,这包括对齐调整,确保分配的内存地址满足特定硬件要求(如对齐到8字节或16字节边界)。 2)空闲…...

深入剖析AI大模型:大模型时代的 Prompt 工程全解析

今天聊的内容&#xff0c;我认为是AI开发里面非常重要的内容。它在AI开发里无处不在&#xff0c;当你对 AI 助手说 "用李白的风格写一首关于人工智能的诗"&#xff0c;或者让翻译模型 "将这段合同翻译成商务日语" 时&#xff0c;输入的这句话就是 Prompt。…...

django filter 统计数量 按属性去重

在Django中&#xff0c;如果你想要根据某个属性对查询集进行去重并统计数量&#xff0c;你可以使用values()方法配合annotate()方法来实现。这里有两种常见的方法来完成这个需求&#xff1a; 方法1&#xff1a;使用annotate()和Count 假设你有一个模型Item&#xff0c;并且你想…...

ardupilot 开发环境eclipse 中import 缺少C++

目录 文章目录 目录摘要1.修复过程摘要 本节主要解决ardupilot 开发环境eclipse 中import 缺少C++,无法导入ardupilot代码,会引起查看不方便的问题。如下图所示 1.修复过程 0.安装ubuntu 软件中自带的eclipse 1.打开eclipse—Help—install new software 2.在 Work with中…...

Springboot社区养老保险系统小程序

一、前言 随着我国经济迅速发展&#xff0c;人们对手机的需求越来越大&#xff0c;各种手机软件也都在被广泛应用&#xff0c;但是对于手机进行数据信息管理&#xff0c;对于手机的各种软件也是备受用户的喜爱&#xff0c;社区养老保险系统小程序被用户普遍使用&#xff0c;为方…...

Android第十三次面试总结(四大 组件基础)

Activity生命周期和四大启动模式详解 一、Activity 生命周期 Activity 的生命周期由一系列回调方法组成&#xff0c;用于管理其创建、可见性、焦点和销毁过程。以下是核心方法及其调用时机&#xff1a; ​onCreate()​​ ​调用时机​&#xff1a;Activity 首次创建时调用。​…...

IP如何挑?2025年海外专线IP如何购买?

你花了时间和预算买了IP&#xff0c;结果IP质量不佳&#xff0c;项目效率低下不说&#xff0c;还可能带来莫名的网络问题&#xff0c;是不是太闹心了&#xff1f;尤其是在面对海外专线IP时&#xff0c;到底怎么才能买到适合自己的呢&#xff1f;所以&#xff0c;挑IP绝对是个技…...

排序算法总结(C++)

目录 一、稳定性二、排序算法选择、冒泡、插入排序归并排序随机快速排序堆排序基数排序计数排序 三、总结 一、稳定性 排序算法的稳定性是指&#xff1a;同样大小的样本 **&#xff08;同样大小的数据&#xff09;**在排序之后不会改变原始的相对次序。 稳定性对基础类型对象…...

2025年渗透测试面试题总结-腾讯[实习]科恩实验室-安全工程师(题目+回答)

安全领域各种资源&#xff0c;学习文档&#xff0c;以及工具分享、前沿信息分享、POC、EXP分享。不定期分享各种好玩的项目及好用的工具&#xff0c;欢迎关注。 目录 腾讯[实习]科恩实验室-安全工程师 一、网络与协议 1. TCP三次握手 2. SYN扫描原理 3. HTTPS证书机制 二…...

Vite中定义@软链接

在webpack中可以直接通过符号表示src路径&#xff0c;但是vite中默认不可以。 如何实现&#xff1a; vite中提供了resolve.alias&#xff1a;通过别名在指向一个具体的路径 在vite.config.js中 import { join } from pathexport default defineConfig({plugins: [vue()],//…...