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 -> { });

    }

}

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

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

相关文章

Apache Calcite 一条 SQL 的查询计划生成之旅【上】

Apache Calcite 一条 SQL 的查询计划生成之旅【上】

作者&#xff1a;沈磊&#xff08;LakeShen&#xff09;&#xff0c;公众号&#xff1a;雷克分析 背景 Apache Calcite 作为一款开源的动态数据管理框架&#xff0c;由于其模块化、可扩展、以及不和任何计算引擎绑定的特性&#xff0c;目前在开源项目和商业化产品中已得到广泛…
阅读更多...
Cesium 自定义Primitive

Cesium 自定义Primitive

1、为什么要自定义Primitive a、在飞机飞行过程中&#xff0c;如果使用entity同时绘制飞机和线的时候&#xff0c;会发现飞机的飞行位置和线的位置不统一&#xff0c;出现差距&#xff0c;出现脱线。 b、结合代码分析&#xff0c;cesium的线和飞机模型是分开渲染的&#xff0c;…
阅读更多...
跨境电商与支付介绍

跨境电商与支付介绍

1、跨境电商定义和分类&#xff1b; 2、国际贸易清结算&#xff1b; 3、跨境支付&#xff1b; 1、跨境电商定义和分类 跨境电商业务简单说就是指不同国家地域的主体通过电子商务进行交易的一种业务模式。同传统的电商不同&#xff0c;交易双方属于不同的国家。因此&#xff0…
阅读更多...
LaMa Image Inpainting 图像修复 ONNX Runtime Demo

LaMa Image Inpainting 图像修复 ONNX Runtime Demo

目录 介绍 效果 模型信息 项目 代码 下载 LaMa Image Inpainting 图像修复 Onnx Demo 介绍 gihub地址&#xff1a;https://github.com/advimman/lama &#x1f999; LaMa Image Inpainting, Resolution-robust Large Mask Inpainting with Fourier Convolutions, WAC…
阅读更多...
C++——友元

C++——友元

目录 友元 友元函数 友元函数使用案例 友元类 友元 友元是C提供的一种突破封装&#xff08;突破类域&#xff09;的方式&#xff0c;有时提供了便利。但是友元会增加耦合度&#xff0c;但破坏了封装&#xff0c;所以友元不宜多用。友元分为友元函数和友元类。 友元函数 友元…
阅读更多...
SpringBoot 的基础使用与入门（总结众多文章与实践）

SpringBoot 的基础使用与入门（总结众多文章与实践）

一&#xff1a;介绍 1、SpringBoot概念 从最根本上来讲&#xff0c;Spring Boot就是一些库的集合&#xff0c;它能够被任意项目的构建系统所使用。简便起见&#xff0c;该框架也提供了命令行界面&#xff0c;它可以用来运行和测试Boot应用。框架的发布版本&#xff0c;包括集成…
阅读更多...
电脑DLL修复工具，一键解决计算机dll丢失

电脑DLL修复工具，一键解决计算机dll丢失

mfc120u.dll 是一个属于 Microsoft Visual Studio 2013 或相关版本的 Microsoft Foundation Classes (MFC) 动态链接库(DLL)文件。MFC 是一个用于简化 Windows 应用程序开发的 C 类库&#xff0c;它封装了许多Windows API 函数&#xff0c;使开发者更容易创建 Windows 应用程序…
阅读更多...
C语言 vs Rust应该学习哪个？

C语言 vs Rust应该学习哪个？

C语言 vs Rust应该学习哪个&#xff1f; 在开始前我有一些资料&#xff0c;是我根据网友给的问题精心整理了一份「C语言的资料从专业入门到高级教程」&#xff0c; 点个关注在评论区回复“888”之后私信回复“888”&#xff0c;全部无偿共享给大家&#xff01;&#xff01;&am…
阅读更多...
Github配置ssh key的步骤

Github配置ssh key的步骤

1. 检查本地主机是否已经存在ssh key 是否存在 id_rsa 和 id_rsa.pub文件&#xff0c;如果存在&#xff0c;说明已经有SSH Key 如下图所示&#xff0c;则表明已经存在 如果存在&#xff0c;直接跳到第三步 2. 生成ssh key 如果不存在ssh key&#xff0c;使用如下命令生…
阅读更多...
奇点云：SAFe框架下，我们对平台软件工程生产线做了4项改造

奇点云：SAFe框架下，我们对平台软件工程生产线做了4项改造

导读&#xff1a; 客户规模扩大&#xff0c;如何保证大数据软件产品和服务质量始终如一&#xff1f;几乎所有成长中的软件厂商&#xff0c;尤其是需要通过私有化部署交付的厂商&#xff0c;都会面临这个问题。正如《人月神话》中多次表明的&#xff0c;单纯地增加人手、扩大团队…
阅读更多...
吴恩达deeplearning.ai:sigmoid函数的替代方案以及激活函数的选择

吴恩达deeplearning.ai:sigmoid函数的替代方案以及激活函数的选择

以下内容有任何不理解可以翻看我之前的博客哦&#xff1a;吴恩达deeplearning.ai专栏 文章目录 引入——改进下需求预测模型ReLU函数(整流线性单元 rectified linear unit&#xff09;线性激活函数(linear activation function)激活函数的选择实现方式为什么需要激活函数 到现在…
阅读更多...
使用SSH推拉Github代码

使用SSH推拉Github代码

快速使用 ssh-keygen -t rsa -b 4096 -C "your_emailexample.com 创建ssh密钥&#xff08;一直回车&#xff0c;不要指定目录&#xff0c;不要设置密码&#xff09;将生成的 ~/id_ras.pub 中的内容复制到Github对应位置即可 1.SSH简介 SSH&#xff08;Secure Shell&…
阅读更多...
Python实现时间序列分析动态因子模型(DynamicFactor算法)项目实战

Python实现时间序列分析动态因子模型(DynamicFactor算法)项目实战

说明&#xff1a;这是一个机器学习实战项目&#xff08;附带数据代码文档视频讲解&#xff09;&#xff0c;如需数据代码文档视频讲解可以直接到文章最后获取。 1.项目背景 动态因子模型&#xff08;Dynamic Factor Models, DFM&#xff09;是一种统计学和计量经济学中用于处理…
阅读更多...
算法项目外包的收费方式

算法项目外包的收费方式

针对算法研究性项目的收费方式和注意事项&#xff0c;这取决于项目的具体性质、规模和所涉及的技术领域。以下是一些常见的收费方式和需要注意的问题&#xff0c;希望对大家有所帮助。北京木奇移动技术有限公司&#xff0c;专业的软件外包开发公司&#xff0c;欢迎交流合作。 收…
阅读更多...
Spring Exception 常见错误

Spring Exception 常见错误

今天&#xff0c;来学习 Spring 的异常处理机制。Spring 提供了一套健全的异常处理框架&#xff0c;以便我们在开发应用的时候对异常进行处理。但是&#xff0c;我们也会在使用的时候遇到一些麻烦&#xff0c;接下来我将通过两个典型的错误案例&#xff0c;带着你结合源码进行深…
阅读更多...
WEB漏洞 SSRF简单入门实践

WEB漏洞 SSRF简单入门实践

一、漏洞原理 SSRF 服务端请求伪造 原理&#xff1a;在某些网站中提供了从其他服务器获取数据的功能&#xff0c;攻击者能通过构造恶意的URL参数&#xff0c;恶意利用后可作为代理攻击远程或本地的服务器。 二、SSRF的利用 1.对目标外网、内网进行端口扫描。 2.攻击内网或本地的…
阅读更多...
Selenium自动化落地实践

Selenium自动化落地实践

01、自动化测试流程图 02、主要过程描述 1、自动化测试的切入点 开展自动化测试的时间点很关键&#xff0c;需要在系统已经过多版本的系统测试&#xff0c;达到稳定之后。 2、可行性分析 在进行项目自动化测试之前&#xff0c;第一步就是要确认其可行性&#xff0c;是否可以…
阅读更多...
物联网与智慧城市的融合：构建智能化、便捷化、绿色化的城市未来

物联网与智慧城市的融合：构建智能化、便捷化、绿色化的城市未来

一、引言 随着科技的飞速发展和城市化的不断推进&#xff0c;物联网技术正逐步渗透到城市的各个领域&#xff0c;成为推动智慧城市建设的核心力量。物联网与智慧城市的融合&#xff0c;不仅为城市治理提供了高效、智能的解决方案&#xff0c;也为市民的生活带来了前所未有的便…
阅读更多...
canvas坐标系统 webgl坐标系统 uv纹理坐标系统 原点

canvas坐标系统 webgl坐标系统 uv纹理坐标系统 原点

一、canvas原点在左上角&#xff0c;x轴正方向向右&#xff0c;y轴正方向向下&#xff0c;一个点对应一个像素 二、webgl原点在正中间&#xff0c;x轴正方向向右&#xff0c;y轴正方向向上&#xff0c;数据显示范围在[-1,1]之间&#xff0c;超过此范围不显示数据 三、uv原点在左…
阅读更多...
【wpf】关于绑定的一点明悟

【wpf】关于绑定的一点明悟

背景简介 软件功能为&#xff0c;读取一个文件夹下的所有子文件夹&#xff0c;每个文件夹对自动对应生成 一组 “按键四个勾选” 按键点击触发&#xff0c;可以发送与其对应文件夹中的一些内容。这个绑定的过程我在之前的文章有过详细的介绍&#xff0c;非常的简单。 这里回顾…
阅读更多...
最新文章

Copyright @ 2022~2023