RabbitMQ的常见工作模式

Work queues 工作队列模式

模式说明

通过Helloworld工程我们已经能够构建一个简单的消息队列的基本项目，项目中存在几个角色:生产 者、消费者、队列，而对于我们真实的开发中 ，对于消息的消费者通过是有多个的。

比如在实现用户注册功能时，用户注册成功，会给响对应用户发送邮件，同时给用户发送手机短信，告诉用户已成功注册网站或者app 应用，这种功能在大部分项目开发中都比较常见 ，而对于helloworld 的应用中虽然能够对 消息进行消费，但是有很大问题: 消息消费者只有一个，当消息量非常大时，单个消费者处理消息就会变得很慢，同时给节点页带来很大压力，导致消息堆积越来越多。对于这种情况，RabbitMQ 提供了工作 队列模式，通过工作队列提供做个消费者，对MQ产生的消息进行消费，提高MQ消息的吞吐率，降低消息的处理时间。处理模型图如下。

在这里插入图片描述

实现步骤

生产者

package cn.wolfcode.java.rabbitmq._02worker;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;/*** Created by wolfcode-fanjialong*/public class NewTask {private static final String TASK_QUEUE_NAME = "task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.142.129");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);String message = "tasequeue";for(int i=0;i<20;i++){channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,(message+i).getBytes("UTF-8"));}System.out.println(" [x] Sent '" + message + "'");}}
}

消费者

package cn.wolfcode.java.rabbitmq._02worker;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.util.concurrent.TimeUnit;/*** Created by wolfcode-fanjialong*/public class Worker {private static final String TASK_QUEUE_NAME = "task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.142.129");final Connection connection = factory.newConnection;final Channel channel = connection.createChannel();channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");//channel.basicQos(1);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}finally {channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });}}
简单问题说明

从结果可以看出消息被平均分配到两个消费方，来对消息进行处理，提高了消息处理效率，创建多个消费者来对消息进行处理。这里RabitMQ采用轮询来对消息进行分发时保证了消息被平均分配到每个消费方 。

但是引入新的问题:真正的生产环境下，对于消息的处理基本不会像我们现在看到的这样，每个消 费方处理的消息数量是平均分配的，比如因为网络原因，机器cpu ，内存等硬问题，消费方处理消息时 同类消息不同机器进行处理时消耗时间也是不一样的，比如1号消费者消费1条消息时1秒，2号消费者消费1条消息是5秒，对于1号消费者比2号消费者处理消息快，那么在分配消息时就应该让1号消费者多收 到消息进行处理，也即是我们通常所说的”能者多劳”,同样Rabbitmq对于这种消息分配模式提供了支持。

问题： 任务量很大，消息虽然得到了及时的消费，单位时间内消息处理速度加快，提高了吞吐量，可 是不同消费者处理消息的时间不同，导致部分消费者的资源被浪费。

解决：采用消息公平分发。

总结：工作队列消息轮询分发消费者收到的消息数量平均分配，单位时间内消息处理速度加快，提高了吞吐量。

工作模式队列-消息公平分发(fair dispatch)

在案例01中对于消息分发采用的是默认轮询分发，消息应答采用的自动应答模式，这是因为当消息进 入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将第n条消息发给第n个消费者。

为了解决这个问题，我们使用 basicQos(prefetchCount = 1) 方法，来限RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。执行模型图如下:

在这里插入图片描述

Pub/Sub 订阅模式

模式说明

在这里插入图片描述

在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）

C：消费者，消息的接收者，会一直等待消息到来

Queue：消息队列，接收消息、缓存消息

Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、

递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

  • Fanout：广播，将消息交给所有绑定到交换机的队列

  • Direct：定向，把消息交给符合指定routing key 的队列

  • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

实现步骤

生产者

package cn.wolfcode.java.rabbitmq._03pubsub;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** Created by wolfcode-fanjialong*/
public class EmitLog {private static final String EXCHANGE_NAME = "logs";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");String message = "info: Hello World!";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}

消费者

package cn.wolfcode.java.rabbitmq._03pubsub;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;/*** Created by wolfcode-fanjialong*/public class ReceiveLogs {private static final String EXCHANGE_NAME = "logs";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, "fanout");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}

Routing 路由模式

模式说明

队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）

消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey

Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息
在这里插入图片描述

图解：

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列

C1：消费者，其所在队列指定了需要 routing key 为 error 的消息

C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息

实现步骤

生产者

package cn.wolfcode.java.rabbitmq._04rounting;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionF![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/6bffe712644346098952dba200523856.png)
actory;
/*** Created by wolfcode-fanjialong*/public class EmitLogDirect {private static final String EXCHANGE_NAME = "direct_logs";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, BuiltinExchangeType.DIRECT);String severity = "info";String message = "directMsg";channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + severity + "':'" + message + "'");}}
}

消费者

package cn.wolfcode.java.rabbitmq._04rounting;import com.rabbitmq.client.*;/*** Created by wolfcode-fanjialong*/public class ReceiveLogsDirect {private static final String EXCHANGE_NAME = "direct_logs";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, BuiltinExchangeType.DIRECT);String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "info");channel.queueBind(queueName, EXCHANGE_NAME, "error");channel.queueBind(queueName, EXCHANGE_NAME, "warning");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}}

Topic 模式

模式介绍

Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型

Exchange 可以让队列在绑定 Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert

在这里插入图片描述

图解：

红色 Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到

黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

实现步骤

生产者

package cn.wolfcode.java.rabbitmq._05topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** Created by wolfcode-fanjialong*/public class EmitLogTopic {private static final String EXCHANGE_NAME = "topic_logs";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");String routingKey = "order1.save";String message = "topicMsg";channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");}}}

消费者

package cn.wolfcode.java.rabbitmq._05topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;/*** Created by wolfcode-fanjialong*/public class ReceiveLogsTopic {private static final String EXCHANGE_NAME = "topic_logs";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 = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "order.*");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}}

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【OpenCV】绘制桌面锁屏时钟

【OpenCV】绘制桌面锁屏时钟

OpenCV 是一个开源的计算机视觉&#xff08;Computer Vision&#xff09;与机器学习软件库&#xff0c;提供了多种图像处理算法与接口&#xff0c;用于解决计算机视觉相关问题。OpenCV 支持多平台与多语言&#xff0c;本文主要记录如何使用 OpenCV-Python 绘制桌面锁屏时钟。 目…
阅读更多...
一、环境配置

一、环境配置

一、下载Ubuntu18.04版本镜像 我的电脑配置比较低(08年奥运限定版哦)&#xff0c;使用的是虚拟机VMware进行安装Ubuntu18.04版&#xff0c;跟书上使用的一样 Ubuntu 18.04镜像 别下载错了哈 二、VMware下安装Ubuntu18.04操作系统 之前写过相关的博文&#xff0c;详细配置可…
阅读更多...
软件测试——性能测试

软件测试——性能测试

目录 一、性能测试简介 二、性能测试指标 三、性能测试的流程 四、Jmeter简介 五、Jmeter常见测试框架 六、Jmeter录制脚本 七、脚本增强 八、Jmeter脚本参数化 九、断言 十、关联 十一、JDBC请求 十二、分布式测试 十三、性能测试报告 一、性能测试简介…
阅读更多...
【LeetCode】一周中的第几天+ 一年中的第几天

【LeetCode】一周中的第几天+ 一年中的第几天

2023-12-30 文章目录 一周中的第几天方法一&#xff1a;模拟思路步骤 方法二&#xff1a;调用库函数方法三&#xff1a;调用库函数 [1154. 一年中的第几天](https://leetcode.cn/problems/day-of-the-year/)方法一&#xff1a;直接计算思路&#xff1a; 方法二&#xff1a;调用…
阅读更多...
【计算机考研】学硕还是专硕？

【计算机考研】学硕还是专硕？

这里面没有什么更好的选择&#xff0c;大家都知道专硕录取人数多&#xff0c;学硕录取人数上&#xff0c;博弈一直存在 单就考研的难度上来说&#xff0c;专硕的难度确实要比学说低很多的&#xff0c;同样的专业课的情况下&#xff0c;专硕的考研数学考的是数学二和英语二&…
阅读更多...
rk3568 gamc0 控制器寄存器配置不了导致连接不上phy

rk3568 gamc0 控制器寄存器配置不了导致连接不上phy

系统启动 GMAC 的驱动开机 log 上出现打印&#xff1a;No PHY found 或者 Cannot attach to PHY。查阅rk 官方gmac 配置指导手册出现改问题的原因可能如下&#xff1a; 但反复检查硬件和软件发现都没问题。看内核启动日志发现gamc0 在启动过程中读取gmac0 的版本id是读取不到&a…
阅读更多...
表格自定义音频播放组件

表格自定义音频播放组件

1.在使用的地方调用 <Audiosvue ref"audio" class"audio-box" :audioFile"item.cont"></Audiosvue> 2.引入组件 import Audiosvue from "/components/audio"; 3.组件 <template><div><div style"d…
阅读更多...
优化Vue项目中 WebStorm:2023.3 对 CSS 和 HTML 的默认注释

优化Vue项目中 WebStorm:2023.3 对 CSS 和 HTML 的默认注释

前言 WebStorm是一种基于JetBrains IntelliJ平台的集成开发环境&#xff08;IDE&#xff09;&#xff0c;专门用于Web开发。它支持JavaScript、HTML和CSS等语言&#xff0c;并提供了丰富的功能和工具&#xff0c;以提高开发人员的效率。但是&#xff0c;在开发Vue项目中发现&a…
阅读更多...
（科目三）计算机操作系统

（科目三）计算机操作系统

1、操作系统的基本概念及功能 1.1概念 一、操作系统的概念 操作系统是运行在裸机之上的&#xff0c;控制和管理计算机硬件和软件资源、合理组织计算机工作流程一级方便用户操作的程序集合。 操作系统由一套分层次的控制程序组成&#xff0c;是计算机硬件的一级扩充是软件系…
阅读更多...
【爬虫逆向实战 逆向滑块 Python+Node】今天逆向的网站有点嘿嘿，还是仅供学习，别瞎搞

【爬虫逆向实战 逆向滑块 Python+Node】今天逆向的网站有点嘿嘿，还是仅供学习，别瞎搞

逆向日期&#xff1a;2024.03.01 使用工具&#xff1a;Node.js、Python 加密方法&#xff1a;AES标准算法 文章全程已做去敏处理&#xff01;&#xff01;&#xff01; 【需要做的可联系我】 AES解密处理&#xff08;直接解密即可&#xff09;&#xff08;crypto-js.js 标准算…
阅读更多...
实现unity场景切换

实现unity场景切换

本文实现两个按键实现场景1和场景2之间的切换 ①首先在unity 3D中创建两个场景&#xff0c;分别为Scene1和Scene2 ②在Scene1中创建一个Button&#xff0c;修改txt内容为“To Scene2”&#xff0c;并在Buttons下创建一个空物体&#xff0c;用于挂载脚本。 脚本Trans Scene.…
阅读更多...
操作系统x面试｜进程与线程

操作系统x面试｜进程与线程

1. 线程进程的区别 进程可以称为是资源分配的最小单元&#xff0c;而线程可以称为是处理器分配的最小单元。 资源包括内存空间。同时进程是一段代码的执行过程&#xff0c;这段代码需要多少的内存在代码确定时已经确定下来了。 处理器就是执行单元&#xff0c;一个进程可以拆解…
阅读更多...
程序员的金三银四求职宝典

程序员的金三银四求职宝典

程序员的金三银四求职宝典 金三银四&#xff0c;即三月和四月&#xff0c;是程序员求职的黄金时期。在这个关键时期&#xff0c;求职者们需要做好充分的准备&#xff0c;以便在面试中脱颖而出。以下是一些建议&#xff0c;帮助程序员在金三银四期间更好地准备求职&#xff1a;…
阅读更多...
【DAY06 软考中级备考笔记】数据结构：树

【DAY06 软考中级备考笔记】数据结构：树

数据结构&#xff1a;树 3月1日 – 天气&#xff1a;晴 之前在B站看的视频讲的是在太过简单&#xff0c;弃了。现在换了新的视频继续&#xff0c;后续会重新看前面的视频补过来。https://www.bilibili.com/video/BV1pT4m1S7uH/ 1. 树的基本概念 需要注意的是&#xff1a; 并不是…
阅读更多...
CentOS7使用yum安装MySQL8.0教程

CentOS7使用yum安装MySQL8.0教程

CentOS7使用yum安装MySQL8.0 一、下载MySQL Yum下载仓库二、安装MySQL三、为MySQL运行提供相关的用户以及环境四、启动MySQL五、MySQL开机自启六、修改MySQL中root账户的密码 一、下载MySQL Yum下载仓库 MySQL下载地址&#xff1a;https://dev.mysql.com/downloads/ 首先进入下…
阅读更多...
Java通过Semaphore控制同一时间只有3个线程运行

Java通过Semaphore控制同一时间只有3个线程运行

怎么控制同一时间只有3个线程运行&#xff1f; 直接上代码 import java.util.Date; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore;public class SemaphoreThreadPoolExample {private static…
阅读更多...
【网站项目】324企业人事信息管理系统

【网站项目】324企业人事信息管理系统

&#x1f64a;作者简介&#xff1a;拥有多年开发工作经验&#xff0c;分享技术代码帮助学生学习&#xff0c;独立完成自己的项目或者毕业设计。 代码可以私聊博主获取。&#x1f339;赠送计算机毕业设计600个选题excel文件&#xff0c;帮助大学选题。赠送开题报告模板&#xff…
阅读更多...
Windows下卸载JDK

Windows下卸载JDK

操作步骤&#xff1a; 直接到windows程序卸载面板进行卸载 然后删除已配置的环境变量
阅读更多...
深入解析Golang的encoding/ascii85库：从基础到实战

深入解析Golang的encoding/ascii85库：从基础到实战

深入解析Golang的encoding/ascii85库&#xff1a;从基础到实战 引言基础知识什么是ASCII85编码&#xff1f;ASCII85编码的工作原理ASCII85编码的优点ASCII85编码的缺点 使用Golang的encoding/ascii85库引入encoding/ascii85包ASCII85编码ASCII85解码实战示例小结 进阶技巧和最佳…
阅读更多...
用于游戏开发的顶级 PYTHON 框架

用于游戏开发的顶级 PYTHON 框架

一、说明 我们试图用python开发游戏&#xff0c;一旦产生这个念头&#xff0c;就伴随这样一个问题&#xff1a;当今用于构建游戏的领先 Python 框架有哪些&#xff1f;python下&#xff0c;支持游戏开发平台有哪些优势&#xff1f;我们在这篇博文中告诉你。 二、高级游戏平台简…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023