消息队列——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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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: 数据
            * */
            @Override
            public 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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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绑定error

        channel.queueBind(queue1Name,exchangeName,"error");

        //队列2绑定error,info,warning
        channel.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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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: 数据
            * */
            @Override
            public 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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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.1
        factory.setPort(5672);              //端口 默认值5672
        factory.setVirtualHost("/itcast");  //设置虚拟机 默认值/
        factory.setUsername("yhy");        //用户名,默认值guest
        factory.setPassword("");     //密码,默认值guest
        //3.创建连接Connection
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel 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: 数据
            * */
            @Override
            public 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);

        //不需要关闭资源
    }
}

 工作模式总结

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/42747.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【itext7】itext7操作PDF文档之添加段落文本内容、添加List列表、添加Image图片、添加Table表格

这篇文章&#xff0c;主要介绍itext7操作PDF文档之添加段落文本内容、添加List列表、添加Image图片、添加Table表格。 目录 一、itext7操作PDF内容 1.1、添加段落文本内容 1.2、添加列表内容 1.3、添加图片 1.4、添加表格 &#xff08;1&#xff09;列宽采用点单位&#…

情绪即需求

情绪即需求 心理学认为&#xff0c;每个情绪背后都藏着一个未被满足的心里需求. 模型介绍 每一个情绪背后&#xff0c;都有一个未被满足的心理需求。情绪没有好坏之分&#xff0c;存在即合理。情绪是人类不断进化的产物&#xff0c;每一种情绪都是在保护我们&#xff0c;都有其…

基于Java+SpringBoot+vue前后端分离校园周边美食探索分享平台设计实现

博主介绍&#xff1a;✌全网粉丝30W,csdn特邀作者、博客专家、CSDN新星计划导师、Java领域优质创作者,博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专…

OpenCv之特征检测

目录 一、基本概念 二、harris角点检测 三、SIFT算法 四、Shi-Tomasi角点检测 一、基本概念 特征检测指的是使用计算机提取图像信息&#xff0c;决定每个图像的点是否属于一个图像特征。特征检测的结果是把图像上的点分为不同的子集&#xff0c;这些子集往往属于孤立的点、…

Element Plus 日期选择器

计算开始日期到结束日期的总天数 结构 <el-form-item label"计划开始时间" required prop"StartTime"><el-date-pickertype"date"v-model"ruleForm.StartTime":disabled-date"StartTime"placeholder"计划开始…

图像处理之LoG算子(高斯拉普拉斯)

LoG算子&#xff08;高斯拉普拉斯算子&#xff09; LoG算子是由拉普拉斯算子改进而来。拉普拉斯算子是二阶导数算子&#xff0c;是一个标量&#xff0c;具有线性、位移不变性&#xff0c;其传函在频域空间的原点为0。所有经过拉普拉斯算子滤波的图像具有零平均灰度。但是该算子…

RT-Thread qemu mps2-an385 bsp 移植制作 :系统运行篇

前言 前面已经让 RT-Thread 进入了 entry 入口函数&#xff0c;并且 调整 链接脚本&#xff0c;自动初始化与 MSH shell 的符号已经预留&#xff0c; 进入了 RT-Thread 的初始化流程 接下来&#xff1a;从 内存管理、系统tick 定时器、适配串口 uart 驱动三个模块入手&#xf…

CPU密集型和IO密集型任务的权衡:如何找到最佳平衡点

关于作者&#xff1a;CSDN内容合伙人、技术专家&#xff0c; 从零开始做日活千万级APP。 专注于分享各领域原创系列文章 &#xff0c;擅长java后端、移动开发、人工智能等&#xff0c;希望大家多多支持。 目录 一、导读二、概览三、CPU密集型与IO密集型3.1、CPU密集型3.2、I/O密…

【字符流】案例:集合到文件

案例&#xff1a;集合到文件 1.需求&#xff1a; 把ArrayList集合中的字符串数据写入到文本文件。要求&#xff1a;每一个字符串元素作为文件中的一行数据 2.思路 创建ArrayList集合往集合中存储字符串元素创建字符缓冲输出流对象遍历集合&#xff0c;得到每一个字符串数据调…

SpringBoot中间件—ORM(Mybatis)框架实现

目录 定义 需求背景 方案设计 代码展示 UML图 实现细节 测试验证 总结 源码地址&#xff08;已开源&#xff09;&#xff1a;https://gitee.com/sizhaohe/mini-mybatis.git 跟着源码及下述UML图来理解上手会更快&#xff0c;拒绝浮躁&#xff0c;沉下心来搞 定义&#x…

界面控件DevExpress WPF数据编辑器组件,让数据处理更灵活!(二)

界面控件DevExpress WPF编辑器库可以帮助用户提供直观的用户体验&#xff0c;具有无与伦比的运行时选项和灵活性。WPF数据编辑器提供了全面的掩码和数据验证支持&#xff0c;可以独立使用&#xff0c;也可以作为容器控件(如DevExpress WPF Grid和WPF TreeList)中的单元格编辑器…

【pytho】request五种种请求处理为空和非空处理以及上传excel,上传图片处理

一、python中请求处理 request.args获取的是个字典&#xff0c;所以可以通过get方式获取请求参数和值 request.form获取的也是个字典&#xff0c;所以也可以通过get方式获取请求的form参数和值 request.data&#xff0c;使用过JavaScript&#xff0c;api调用方式进行掺入jso…

Cmake笔记记录

工作后开发内容都是在Linux系统下完成&#xff0c;cmake使用比较频繁&#xff0c;找到一篇很不错的CMake笔记。 记录下来方便自己查阅。 目录 1.CMake介绍2.示例一&#xff1a;编译单个文件CMAKE_BUILD_TYPE 3.示例二&#xff1a;编译同个目录下多个文件aux_source_directory()…

QT通过TCP协议发送结构体数据

文章目录 前言1. memcpy方式1.1 发送整个结构体1.2 发送部分数据 2. QDataStream2.1 符号<<2.2 wrieteRawData 总结 前言 做上位机的时候下发数据&#xff0c;数据是一个结构体&#xff0c;这里就阐述一下怎么通过TCP协议发送结构体。 我自己写的时候主要通过两种方式&a…

JavaWeb 速通HTTP

目录 一、HTTP快速入门 1.HTTP简介 : 2.HTTP请求头 : 3.HTTP响应头 : 二、HTTP响应状态码 1.基本介绍 : 2.常见状态码 : 3.状态码的分类 : 4.完整状态码汇总 : 三、HTTP请求包和响应包 1.请求包分析 : 1 GET请求 (1) 说明 (2) doGet返回数据给浏览器 (3) form表单提…

华为云CodeArts产品体验的心得体会及想法

文章目录 前言CodeArts 的产品优势一站式软件开发生产线研发安全Built-In华为多年研发实践能力及规范外溢高质高效敏捷交付 功能特性说明体验感受问题描述完结 前言 华为云作为一家全球领先的云计算服务提供商&#xff0c;致力于为企业和个人用户提供高效、安全、可靠的云服务。…

C++---树形DP---树的中心(每日一道算法2023.7.19)

注意事项&#xff1a; 本题为"树形DP—树的最长路径"的近似题&#xff0c;同时涉及到 单链表模拟邻接表存储图 的操作&#xff0c;建议先理解那篇文章。 题目&#xff1a; 给定一棵树&#xff0c;树中包含 n 个结点&#xff08;编号1~n&#xff09;和 n−1 条无向边…

uniapp查看ios打包后的Info.plist文件

最近在用 uni 开发 ios 的时候给项目添加了自定义的 Info.plist 文件&#xff0c;但是打包后发现并没有生效&#xff0c;才有了查看打包后的 Info.plist 文件想法。 HBuilderX3.6.5起&#xff0c;支持直接在应用项目中配置 iOS 平台的 Info.plist 和 资源文件&#xff08;Bundl…

C# List 详解二

目录 5.Clear() 6.Contains(T) 7.ConvertAll(Converter) ,toutput> 8.CopyTo(Int32, T[], Int32, Int32) 9.CopyTo(T[]) 10.CopyTo(T[], Int32) C# List 详解一 1.Add(T)&#xff0c;2.AddRange(IEnumerable)&#xff0c;3.AsReadOnly()&…

RabbitMQ集群搭建

说明&#xff1a;集群&#xff0c;不管是Redis集群&#xff0c;还是MQ集群&#xff0c;都是为了提高系统的可用性&#xff0c;使系统不至于因为Redis、MQ宕机而崩溃。本文介绍RabbitMQ集群搭建&#xff0c;RabbitMQ集群分为以下三类&#xff1a; 普通集群 镜像集群 仲裁队列 …
最新文章