.Net RabbitMQ(消息队列)

文章目录

  • 一.RabbitMQ 介绍以及工作模式
    • 1.RabbitMQ的介绍:
    • 2.RabbitMQ的工作模式:
  • 二.RabbitMQ安装
    • 1.安装Erlang语言环境
    • 2.安装RabbitMQ
  • 三.在.Net中使用RabbitMQ
    • 1.HelloWorld模式
    • 2.工作队列模式
    • 3.发布订阅模式
    • 4.Routing路由模式和Topics通配符模式


一.RabbitMQ 介绍以及工作模式

1.RabbitMQ的介绍:

RabbitMQ 是一个开源的消息代理软件,它实现了高级消息队列协议(AMQP)的标准,同时也支持其他消息协议,如 MQTT。它最初由 Rabbit Technologies 公司开发,后来成为了 Pivotal Software 的一部分,现在则是由 VMware 管理。

以下是 RabbitMQ 的一些关键特点和功能:

消息队列: RabbitMQ 允许应用程序之间通过发送和接收消息进行通信。它作为一个中间件,负责接收、存储和路由消息,以确保消息能够可靠地传递到目标应用程序。

灵活的消息路由: RabbitMQ 支持不同类型的交换机(exchange),如 direct、topic、fanout 和 headers,每种类型的交换机都有不同的消息路由方式,使得消息的分发更加灵活高效。

持久化: RabbitMQ 支持将消息持久化到磁盘,以确保即使在代理重启后,消息也不会丢失。它还允许将队列和交换机设置为持久化,以提高系统的可靠性。

集群和高可用性: RabbitMQ 支持构建具有高可用性和可伸缩性的集群。通过将多个 RabbitMQ 节点组成集群,可以实现消息的负载均衡和故障转移,从而提高系统的可靠性和性能。

消息确认机制: RabbitMQ 支持消息确认机制,即生产者可以选择在消息发送到队列后等待消费者对消息的确认,以确保消息被成功处理。这种机制可以保证消息不会丢失,并且可以防止消息重复处理。

插件系统: RabbitMQ 提供了丰富的插件系统,可以通过插件扩展其功能,如集成其他消息协议、实现认证和授权、监控和管理等。

多语言客户端: RabbitMQ 提供了多种编程语言的客户端库,如 Java、Python、Ruby、C# 等,使得开发者可以方便地与 RabbitMQ 进行交互。

总的来说,RabbitMQ 是一个功能强大、可靠性高的消息代理软件,广泛应用于各种分布式系统和应用场景,如微服务架构、实时数据处理、任务调度等。

2.RabbitMQ的工作模式:

RabbitMQ 的工作模式主要涉及以下几个关键组件:生产者、交换机(Exchange)、队列(Queue)和消费者。通过这些组件,RabbitMQ 实现了消息的传输和路由。

生产者(Producer): 生产者是消息的发送者,它负责将消息发送到 RabbitMQ 中。生产者通常将消息发送到一个特定的交换机,然后由交换机将消息路由到一个或多个队列中。

交换机(Exchange): 交换机是消息的路由中心,负责接收从生产者发送的消息,并根据指定的路由规则将消息路由到一个或多个队列中。RabbitMQ 提供了不同类型的交换机,如 direct、topic、fanout 和 headers,每种类型的交换机都有不同的路由规则。

队列(Queue): 队列是消息的存储容器,它负责存储从交换机接收到的消息,以便消费者可以按顺序处理消息。每个队列都有一个唯一的名称,并且可以有多个消费者订阅同一个队列。

消费者(Consumer): 消费者是消息的接收者,它负责从队列中获取消息并进行处理。消费者通常会订阅一个或多个队列,并在收到消息后执行相应的业务逻辑。

RabbitMQ 的工作流程可以简单描述为:

生产者将消息发送到一个特定的交换机。
交换机根据指定的路由规则将消息路由到一个或多个队列中。
消费者从队列中获取消息并进行处理。
在这个过程中,RabbitMQ 提供了丰富的路由和消息确认机制,以确保消息能够可靠地传递和处理。

二.RabbitMQ安装

bilibili三分钟RabbitMQ安装教程,亲测有效

1.安装Erlang语言环境

a.RabbitMQ是Erlang语言开发的,所以我们首先去Erlang官网下载Erlang环境

在这里插入图片描述

b.下载完成后,我们直接根据默认的情况安装到电脑上,然后进入windows的“高级系统设置”,设置Erlang环境变量

给他一个变量名称,然后浏览刚才安装的Erlang目录位置,获取变量值

在这里插入图片描述

c.在环境变量的path中,配置刚才配置好的Erlang环境

在这里插入图片描述

2.安装RabbitMQ

a.接着我们进入RabbitMQ官网下载RabbitMQ

在这里插入图片描述
b.RabbitMQ下载安装完毕后,我们进入RabbitMQ的sbin目录中,用cmd终端行打开命令行,分别执行以下命令安装RabbitMQ Cli

a.在cmd中,进入RabbitMQ的sbin目录下,输入 
rabbitmq-plugins enable rabbitmq_management 
命令安装RabbitMQ Ctl

b.输入 
rabbitmqctl status 
命令检验是否安装成功

c.进入服务中,先停止RabbitMQ的服务,然后在RabbitMQ的sbin目录中,启动RabbitMQ.Server.bat文件,最后再重启RabbitMQ服务,就可以进入浏览器访问http://localhost:15672/进入RabbitMQ服务端了

1)停止rabbitMQ服务

在这里插入图片描述

2)启动rabbitmq.server.bat文件

在这里插入图片描述

3)重启RabbitMQ服务后,使用浏览器进入http://localhost:15672/
斜体样式
默认的账户名密码都是:guest
在这里插入图片描述

三.在.Net中使用RabbitMQ

bilibili某博主详细解说:RabbitMQ从零到高可用集群

1.HelloWorld模式

HelloWorld模式通常被视为RabbitMQ的点对点模式,生产者将消息放入队列中,消费者直接从队列中读取

a.生产者代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        //创建队列,声明并创建一个队列,如果队列已存在,则使用这个队列
        //第一个参数:队列名称ID
        //第二个参数:是否持久化,false对应不持久化数据,MQ停掉数据就会丢失
        //第三个参数,是否队列私有化,flase达标所有的消费则会都可以访问,true代表只有第一次拥有它的消费者才能一直使用
        //第四个参数:是否自动删除,false代表链接停掉后不自动删除这个队列

        channel.QueueDeclare("hello",true,false,false,null);
        string message = $"Hellow RabbitMQ-{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}";
        var body = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish("","hello",null,body);
        Console.WriteLine($"RabbitMQ生产者已发送{message}");
        Console.ReadKey();
    }
}

b.消费者代码:

 var factory = new ConnectionFactory();
 factory.HostName = "localhost";
 factory.Port = 5672;
 factory.UserName = "guest";
 factory.Password = "guest";

 using (var connection = factory.CreateConnection())
 {
     using (var channel = connection.CreateModel())
     {
         //创建一个名为hello的消息队列
         channel.QueueDeclare("hello",true,false,false,null);
         var consumer = new EventingBasicConsumer(channel);
         channel.BasicConsume("hello",false,consumer);

         consumer.Received += (model, e) =>
         {
             var body = e.Body.ToArray();
             var message = Encoding.UTF8.GetString(body);

             Console.WriteLine("RabbitMQ消费者接收到消息:" + message);
         };

         channel.BasicConsume("hello",false,consumer);
         Console.WriteLine("Press [Enter] to exit");
         Console.ReadKey();

     }
 }

c.运行结果:

在这里插入图片描述

2.工作队列模式

工作队列模式,也称任务队列模式,生产者将任务传递到队列中,多个消费者竞争消费队列中的消息,一条消息只对应一个消费者

a.生产者代码:

 var factory = new ConnectionFactory();
 factory.HostName = "localhost";
 factory.Port = 5672;
 factory.UserName = "guest";
 factory.Password = "guest";

 using (var connection = factory.CreateConnection() )
 {
     using (var channel = connection.CreateModel())
     {
         channel.QueueDeclare("WorkQueue",true,false,false,null);
         for (int i = 0; i < 10; i++)
         {
             var msg = $"消息队列ID10000{i}发送成功,{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss:fff")}";
             var body = Encoding.UTF8.GetBytes(msg);
             channel.BasicPublish("","WorkQueue",null,body);
             Console.WriteLine($"已发送:{msg}");
         }
         Console.ReadKey();
     }
 }

b.消费者1代码:

 var factory = new ConnectionFactory();
 factory.HostName = "localhost";
 factory.Port = 5672;
 factory.UserName = "guest";
 factory.Password = "guest";

 using (var connection = factory.CreateConnection())
 {
     using (var channel = connection.CreateModel())
     {
         //创建一个名为hello的消息队列
         channel.QueueDeclare("WorkQueue", true, false, false, null);
         //basicqos,MQ不再对消费者一次发送多条请求,而是一条处理完毕(确然后)再从队列中获取新的
         channel.BasicQos(0, 1, false);
         var consumer = new EventingBasicConsumer(channel);

         channel.BasicConsume("WorkQueue", false, consumer);

         consumer.Received += (model, e) =>
         {
             var body = e.Body.ToArray();
             var message = Encoding.UTF8.GetString(body);
             Thread.Sleep(100);
             Console.WriteLine("RabbitMQ消费者接收到消息:" + message);
             //消息确认:一旦消费者成功处理了任务,它会发送确认消息给 RabbitMQ,告知 RabbitMQ 该消息已经被处理并可以从队列中删除。如果消费者在处理任务时发生错误,消息将会被重新放回队列中,以便其他消费者重新尝试处理。
             channel.BasicAck(e.DeliveryTag,false);
         };

         channel.BasicConsume("WorkQueue", false, consumer);
         Console.WriteLine("Press [Enter] to exit");
         Console.ReadKey();

     }
 }

c.消费者2代码:

与消费者1代码一致

d.运行结果:

在这里插入图片描述

3.发布订阅模式

发布/订阅(Publish/Subscribe)模式是一种消息传递模式,用于将消息广播给多个消费者。在这种模式下,生产者将消息发送到一个交换机(Exchange)中,而不是直接发送到队列中。交换机则将消息广播给与之绑定的多个队列,每个队列都有一个或多个消费者来接收消息。

a.生产者代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        var msg = "20摄氏度";
        var body = Encoding.UTF8.GetBytes(msg);
        channel.BasicPublish("aweather","",null,body);
        Console.WriteLine("天气发送成功:"+msg);
    }
}

b.消费者1代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare("aweather", ExchangeType.Fanout);
        //声明消息队列
        channel.QueueDeclare("HuBei", true, false, false, null);
        channel.QueueBind("HuBei", "aweather", "");
        var consumer = new EventingBasicConsumer(channel);

        consumer.Received += (model, e) =>
        {
            var msg = Encoding.UTF8.GetString(e.Body.ToArray());
            Console.WriteLine("湖北收到天气信息:" + msg);
            channel.BasicAck(e.DeliveryTag, false);
        };

        channel.BasicConsume("HuBei", false, consumer);
        Console.WriteLine("Press [Enter] to exit");
        Console.ReadKey();

    }
}

c.消费者2代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare("aweather",ExchangeType.Fanout);
        //生命消息队列
        channel.QueueDeclare("HuNan",true,false,false,null);
        channel.QueueBind("HuNan","aweather","");
        var consumer = new EventingBasicConsumer(channel);

        consumer.Received += (model, e) =>
        {
            var msg = Encoding.UTF8.GetString(e.Body.ToArray());
            Console.WriteLine("湖南收到天气信息:"+msg);
            channel.BasicAck(e.DeliveryTag,false);
        };

        channel.BasicConsume("HuNan",false,consumer);
        Console.WriteLine("Press [Enter] to exit");
        Console.ReadKey();

    }
}

d.运行结果:

在这里插入图片描述

4.Routing路由模式和Topics通配符模式

Routing路由模式,是发布订阅模式的延伸,在发布订阅模式的基础上,分配
路由键(Routing Key),消费者根据路由键匹配队列中对用的消息

Topics通配符模式,是Routing路由模式的延伸,*用于匹配一个单词,#用于匹配零个或多个单词,用于替换匹配路由键

注意:一个队列只能匹配一个通配符,但是可以匹配多个路由规则,路由模式和通配符模式可以嵌套使用

a.生产者代码:

var dic = new Dictionary<string, string>();
dic.Add("china.hunan.changsha.20240418","中国湖南长沙2024年04月18日天气数据");
dic.Add("china.hubei.wuhan.20240418", "中国湖北武汉2024年04月18日天气数据");
dic.Add("china.beijing.20240418", "中国北京2024年04月18日天气数据");
dic.Add("us.flld.20240418", "美国弗罗里达2024年04月18日天气数据");

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        foreach (var item in dic)
        {
            channel.BasicPublish("weather",item.Key,null,Encoding.UTF8.GetBytes(item.Value));
        }
        Console.WriteLine("气象信息发送成功");
    }
}

b.消费者1代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare("weather",ExchangeType.Direct);
        channel.QueueDeclare("us",true,false,false,null);
        //通配符模式
        channel.QueueBind("us","weather", "us.*.20240418");
        //路由模式
        channel.QueueBind("us","weather", "china.beijing.20240418");
        channel.BasicQos(0, 1, false);
        var customer = new EventingBasicConsumer(channel);
        
        customer.Received += (model, e) =>
        {
            var msg = Encoding.UTF8.GetString(e.Body.ToArray());
            Console.WriteLine("美国收到天气信息:"+msg);
            channel.BasicAck(e.DeliveryTag,false);
        };

        channel.BasicConsume("us",false,customer);
        Console.WriteLine("Press [Enter] to exit");
        Console.Read();
    }
}

c.消费者2代码:

var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
factory.UserName = "guest";
factory.Password = "guest";
using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare("weather",ExchangeType.Direct);
        channel.QueueDeclare("china",true,false,false,null);
        //路由模式
        channel.QueueBind("china","weather", "china.hunan.changsha.20240418");
        channel.QueueBind("china", "weather", "china.hubei.wuhan.20240418");
        channel.QueueBind("china", "weather", "china.beijing.20240418");
        //通配符模式
        //channel.QueueBind("china", "weather", "china.#");
        channel.BasicQos(0,1,false);

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, e) =>
        {
            var msg = Encoding.UTF8.GetString(e.Body.ToArray());
            Console.WriteLine("中国收到天气消息:"+msg);
            channel.BasicAck(e.DeliveryTag,false);
        };
        channel.BasicConsume("china",false,consumer);
        Console.WriteLine("Press [Enter] to exit");
        Console.Read();
    }
}

d.运行结果:

在这里插入图片描述

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

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

相关文章

安全开发实战(4)--whois与子域名爆破

目录 安全开发专栏 前言 whois查询 子域名 子域名爆破 1.4 whois查询 方式1: 方式2: 1.5 子域名查询 方式1:子域名爆破 1.5.1 One 1.5.2 Two 方式2:其他方式 总结 安全开发专栏 安全开发实战​​http://t.csdnimg.cn/25N7H 前言 whois查询 Whois 查询是一种用…

java.lang.OutOfMemoryError: WrappedJavaFileObject --idea启动项目内存溢出解决

java.lang.OutOfMemoryError 解决方案 现象 项目开发时&#xff0c;启动idea&#xff0c;报内存溢出错误&#xff0c;如下&#xff1a; java: java.lang.OutOfMemoryError: WrappedJavaFileObject.....解决 通过 调整idea 的 配置参数 来调整 jvm 大小解决。 -Xmx8192m-Xm…

C++进修——C++基础入门

初识C 书写HelloWorld #include <iostream> using namespace std;int main() {cout << "HelloWorldd" << endl;system("pause");return 0; }注释 作用&#xff1a;在代码中加一些说明和解释&#xff0c;方便自己或其他程序员阅读代码…

二分法问题

日升时奋斗&#xff0c;日落时自省 目录 1、二分法 2、二分法问题 2.1 、在排序数组中查找元素的第一个和最后一个位置 2.2、搜索插入位置 2.3、山脉数组的峰顶索引 2.4、0-n-1中缺失的数字 1、二分法 二分法是比较简单的一种查找算法&#xff0c;但是效率很高&#xff0…

【创建型模式】原型模式

一、原型模式概述 原型&#xff08;Prototype&#xff09;模式的定义&#xff1a;用一个已经创建的实例作为原型&#xff0c;通过复制该原型对象来创建一个和原型相同或相似的新对象。在这里&#xff0c;原型实例指定了要创建的对象的种类。用这种方式创建对象非常高效&#xf…

【Qt】Qt安装包、源码、子模块(submodules)下载

1、Qt 4.0 ~ Qt5.14 Qt 4.0 ~ Qt5.14 离线安装包、源码和子模块(submodules)源码下载路径: https://download.qt.io/new_archive/qt/以Qt5.7.1为例,注意子模块都是源码,需要独立编译 2、Qt5.15 ~ Qt6.7 Qt5.15 ~ Qt6.7源码和子模块(submodules)源码下载路径: htt…

分类算法——决策树(五)

认识决策树 决策树思想的来源非常朴素&#xff0c;程序设计中的条件分支结构就是if-else结构&#xff0c;最早的决策树就是利用这类结构分割数据的一种分类学习方法。 决策树分类原理详解 为了更好理解决策树具体怎么分类的&#xff0c;通过一个问题例子&#xff1a; 问题…

【MIT6.824】lab3 Fault-tolerant Key/Value Service 实现笔记

引言 lab3A的实验要求如下&#xff1a; Your first task is to implement a solution that works when there are no dropped messages, and no failed servers. You’ll need to add RPC-sending code to the Clerk Put/Append/Get methods in client.go, and implement Pu…

HiveSql中的函数家族(二)

一、窗口函数 1、什么是窗口函数 在 SQL 中&#xff0c;窗口函数&#xff08;Window Functions&#xff09;是一种特殊的函数&#xff0c;它允许在查询结果集的特定窗口&#xff08;通常是一组行&#xff09;上执行聚合、分析和计算操作&#xff0c;而无需聚合整个结果集。窗口…

使用Python工具库SnowNLP对评论数据标注(二)

这一次用pandas处理csv文件 comments.csv import pandas as pd from snownlp import SnowNLPdf pd.read_csv("C:\\Users\\zhour\\Documents\\comments.csv")#{a: [1, 2, 3], b: [4, 5, 6], c: [7, 8, 9]}是个字典 emotions[] for txt in df[sentence]:s SnowNLP(…

接收区块链的CCF会议--ICSOC 2024 截止7.24

ICSOC是CCF B类会议&#xff08;软件工程/系统软件/程序设计语言&#xff09; 2023年长文短文录用率22% Focus Area 4: Emerging Technologies Quantum Service Computing Digital Twins 3D Printing/additive Manufacturing Techniques Blockchain Robotic Process Autom…

【QT+OpenCV】车牌号检测 学习记录 遇到的问题

【QTOpenCV】车牌号检测 学习记录 首先在QT里面配置好OpenCV .pro文件中加入&#xff1a; INCLUDEPATH G:/opencv/build/include LIBS -L"G:/opencv/build/x64/vc14/lib"\-lopencv_core \-lopencv_imgproc \-lopencv_highgui \-lopencv_ml \-lopencv_video \-lo.c…

Meta Llama 3强势来袭:迄今最强开源大模型,性能媲美GPT-4

前言 Meta的最新语言模型Llama 3已经发布&#xff0c;标志着在大型语言模型&#xff08;LLM&#xff09;领域的一次重大突破&#xff0c;其性能在行业内与GPT-4相媲美。此次更新不仅提升了模型的处理能力和精确性&#xff0c;还将开源模型的性能推向了一个新的高度。 Huggingf…

Docker八股总结

1. 容器和虚拟机的区别 传统虚拟机技术是虚拟出一套硬件后&#xff0c;在其上运行一个完整操作系统&#xff0c;在该系统上再运行所需应用进程&#xff1b;而容器内的应用进程直接运行于宿主的内核&#xff0c;容器内没有自己的内核&#xff0c;而且也没有进行硬件虚拟。因此容…

2021年全国大学生电子设计竞赛D题——基于互联网的摄像测量系统(二)

09 电路设计 前面介绍了系统的硬件框图如下&#xff1a; 硬件基本分为三块&#xff0c;两个摄像节点&#xff0c;一个终端节点。 1. 摄像节点硬件 摄像节点由一个DE10-Nano开发板和一个D8M摄像头实现&#xff0c;DE10-Nano开发板的HDMI接口外接HDMI显示器来显示拍摄到的视频。…

Flask + Bootstrap vs Flask + React/Vue:初学者指南

在这篇博客文章中&#xff0c;我们将比较 Flask Bootstrap 和 Flask React/Vue 这两种技术栈&#xff0c;以帮助初学者了解哪种组合更适合他们的项目需求。我们将从学习曲线、易用性、依赖管理、构建部署和路由定义等方面进行比较。 学习曲线 Flask 是一个基于 Python 的轻…

信息系统项目管理师0055:优化和持续改进(4信息系统管理—4.1管理方法—4.1.5优化和持续改进)

点击查看专栏目录 文章目录 4.1.5优化和持续改进1.定义阶段2.度量阶段3.分析阶段4.改进/设计阶段5.控制/验证阶段4.1.5优化和持续改进 优化和持续改进是信息系统管理活动中的一个环节,良好的优化和持续改进管理活动能够有效保障信息系统的性能和可用性等,延长整体系统的有效使…

偏微分方程算法之一阶双曲差分法

目录 一、研究目标 二、理论推导 2.1 引言 2.2 迎风格式 2.3 完全不稳定差分格式 2.4 蛙跳格式&#xff08;Leapfrog&#xff09; 2.5 Lax-Friedrichs格式 2.6 Lax-Wendroff格式 2.7 Beam-Warming格式 2.8 隐格式 2.9 Courant-Friedrichs-Lewy条件&#xff08;CFL条…

一文学会时序约束

主时钟约束命令/生成时钟约束命令IO输入输出延迟约束命令及效果最大最小延迟命令及作用多周期路径怎么约束什么情况设置伪路径时钟组设置的三个选项 如果不了解时序分析可以先看下下面这篇文章&#xff1a; 数字IC/FPGA——时序分析 目录 1.时钟约束&#xff08;1&#xff09;…

线性代数---行列式的性质

1. 行列式的行与列(按原顺序)互换