如何用.NETCore操作RabbitMQ
作者：青城同学 发布时间：2022-06-20 04:11:04
什么是RabbitMQ？
RabbitMQ是由erlang语言开发的一个基于AMQP(Advanced Message Queuing Protocol)协议的企业级消息队列中间件。可实现队列，订阅/发布，路由，通配符等工作模式。
为什么要使用RabbitMQ?
异步处理：比如发送邮件，发送短信等不需要等待处理结果的操作
应用解耦：比如下单成功后，通知仓库发货，不需要等待仓库回应，通过消息队列去通知仓库，降低应用间耦合程序，可并行开发两个功能模块
流量削锋：在抢购或者其他的活动页，服务处于爆发式请求状态，如果直连数据库，数据库容易被拖垮。抢购商品也容易出现库存超卖的情况。通过队列可有效解决该问题。
日志处理：在单机中，日志直接写入到文件目录中，但是在分布式应用中，日志需要有统一的处理机制，可通过消息队列统一由某个消费端做处理。
消息通信：如生产端和消费端可通过队列进行异步通信
如何安装RabbitMQ?
Windows端
1.安装erlang语言运行环境
https://erlang.org/download/otp_win64_23.2.exe
下载后直接下一步即可
2.安装RabbitMQ
https://www.rabbitmq.com/install-windows.html
直接点击安装下一步即可按章
3.安装RabbitMQ的Web管理平台
RabbitMQ的管理平台是通过插件的形式使用，需要手动启用管理平台
在Windows下，RabbitMQ默认被安装到C:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.14 下。
打开sbin ，在cmd或者powershell中执行
rabbitmq-plugins.bat enable rabbitmq_management
安装完成后，浏览器打开 http://localhost:15672/ 即可看到RabbitMQ的管理界面。输入默认账号密码 guest 成功登录。
Linux环境安装
1.Ubuntu:https://www.rabbitmq.com/install-debian.html
2.Centos:https://www.rabbitmq.com/install-rpm.html
RabbitMQ的基本概念
生产者
发送消息的端
消费者
获取消息并处理的端
Connection
一个终端连接。每一个Connection都可以在RabbitMQ后台看到
Channel
Channel是建立在Connection上的一个虚拟通信管道。一般情况下，往消息队列中写入多条消息，为了不每条消息都建立一个TCP连接，所以RabbitMQ的做法是多条消息可以公用一个Connection，大大提高MQ的负载能力。
Exchange
Exchange是一个虚拟交换机。每一条消息都必须要通过交换机才能能进入对应的队列，可以理解为网络设备中的交换机，是一个意思。
Queue
Queue是一个存储消息的内部对象，所有的Rabbit MQ消息都存储在Queue中。生产者所生产的消息会存储在Queue中，消费者获取的消息也是从Queue中获取。
如何在.NET Core中使用RabbitMQ?
nuget安装
dotnet add package RabbitMQ.Client
创建生产者
const string QUEUENAME = "HELLO_MQ";
//创建连接对象工厂
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672, //RabbitMQ默认的端口
};
while (true)
{
using var conn = factory.CreateConnection();
var chanel = conn.CreateModel();
chanel.QueueDeclare(QUEUENAME, true, false, false);
Console.WriteLine("输入生产内容：");
var input = Console.ReadLine();
chanel.BasicPublish("", QUEUENAME, null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
}
在循环中，输入一个值，按下enter，即可推送一条消息到队列。
也可以直接在RabbitMQ的管理后台查看
可以看到我们发送的消息已经被RabbitMQ存储在Queue中了。只等某个幸运的消费者前来消费。
创建消费者
const string QUEUENAME = "HELLO_MQ";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var chanel = conn.CreateModel();
chanel.QueueDeclare(QUEUENAME, true, false, false);
EventingBasicConsumer consumer = new EventingBasicConsumer(chanel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
chanel.BasicAck(e.DeliveryTag, true); //收到回复后，RabbitMQ会直接在队列中删除这条消息
};
chanel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("启动成功");
Console.ReadLine();
启动成功后，consumer的Received方法，会收到一条来自MQ的消息，
如果处理完成后，不调用chennel的BasicAck方法，那么这条消息依然会存在，下次有消费者出现，会再次推送给消费者。
简单的RabbitMQ Hello World到这里就算完成了。接下来就是稍微高级一点的应用
RabbitMQ的工作模式
Work Queue 工作队列模式
工作队列模式的意思就是一个生产者对应多个消费者。RabbitMQ会使用轮询去给每个消费者发送消息。
publish/subscribe
发布订阅模式是属于比较用多的一种。
发布订阅，是由交换机发布消息给多个队列。多个队列再对应多个消费者。
发布订阅模式对应的交换机类型的fanout。
消费者
A
const string QUEUENAME = "HELLO_MQ_B";
const string TESTEXCHANGE = "TESTEXCHANGE";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var channel = conn.CreateModel();
//定义队列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
//绑定队列到交换机
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
channel.BasicAck(e.DeliveryTag, true); //收到回复后，RabbitMQ会直接在队列中删除这条消息
};
channel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("启动成功");
Console.ReadLine();
B
const string QUEUENAME = "HELLO_MQ";
const string TESTEXCHANGE = "TESTEXCHANGE";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var channel = conn.CreateModel();
//定义队列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
//绑定队列到交换机
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
channel.BasicAck(e.DeliveryTag, true); //收到回复后，RabbitMQ会直接在队列中删除这条消息
};
channel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("启动成功");
Console.ReadLine();
生产者
const string QUEUENAME = "HELLO_MQ";
const string QUEUENAME_B = "HELLO_MQ_B";
const string TESTEXCHANGE = "TESTEXCHANGE";
//创建连接对象工厂
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672, //RabbitMQ默认的端口
};
using var conn = factory.CreateConnection();
while (true)
{
var channel = conn.CreateModel();
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
Console.WriteLine("输入生产内容：");
var input = Console.ReadLine();
channel.BasicPublish(TESTEXCHANGE,"", null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
}
在生产者运行成功后，RabbitMQ后台会出现一个交换机，点击交换机会看到交换机下绑定了两个队列
从生产者发送消息到队列，两个消费者会同时收到消息
routing模式
routing模式对应的交换机类型是direct，和发布订阅模式的区别在于：routing模式下，可以指定一个routingkey，用于区分消息
生产者
var channel = conn.CreateModel();
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false);
//绑定队列到交换机
Console.WriteLine("输入生产内容：");
var input = Console.ReadLine();
channel.BasicPublish(TESTEXCHANGE, "INFO", null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
消费者 A
//定义队列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false);
//绑定队列到交换机
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "INFO");
消费者 B
//定义队列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定义交换机
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false);
//绑定队列到交换机
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "ERROR");
绑定成功后，发送消息，消费者A可以收到消息，消费者B无法收到消息。
如果遇到指定routingKey生产一条消息，结果 AB消费者都收到的情况。建议在RabbitMQ后台的交换机下看一下绑定的Queue是否重复绑定了多个routingKey.
topic通配符模式
在通配符模式下，RabbitMQ使用模糊匹配来决定把消息推送给哪个生产者。通配符有两个符号来匹配routingKey
1.*匹配一个字符 如:*.qq.com 可匹配 1.qq.com
2.#匹配一个或者多个字符。 如：*.qq.com 可匹配 1.qq.com或者1111.qq.com
其他的操作基本和routing模式一样。
header模式
header模式是把routingkey放到header中.取消掉了routingKey。并使用一个字典传递 K、V的方式来匹配。
比如同时要给用户发送邮件和短信，可直接通过header的键值对来匹配绑定的值，把消息传递给发短信和邮件的生产者.
来源：https://www.cnblogs.com/boxrice/p/14475574.html
0
投稿猜你喜欢
1.docker安装seata 1.3.0镜像docker pull seataio/seata-server:1.3.02.运行容器获取配- 需求：视频通话界面，两个surfaceView一个显示本端的视图，另一个显示对端的视图，由于显示比例的问题总会存在一个覆盖另一个的问题，为保
- 本文实例讲述了Android开发使用HttpURLConnection进行网络编程。分享给大家供大家参考，具体如下：——HttpURLCon
- 一、什么是锁擦除锁擦除是指虚拟机即时编译器（JIT）在运行时，对一些代码上要求同步，但是被检测到不可能存在共享数据竞争的锁进行擦除。锁擦除的
- 上代码喽~package ncu.com.app.chatpter_5;import java.util.Random;//结点类class
- 本文实例为大家分享了UnityShader3实现彩光效果展示的具体代码，供大家参考，具体内容如下参考链接： 【OpenGL】Shader实例
- 1. 异常1.1 try…catch异常处理try catch的异常处理的格式写法 :try{ &nbs
- 本文实例讲述了asp.net实现遍历Request的信息操作。分享给大家供大家参考，具体如下：#需求：在服务端获取从客户端发送过来的所有数据
- 先给大家展示下效果图，对第三方开源 android tickplusdrawable相关知识感兴趣的朋友一起学习吧。Android tick
- NameServer1.架构设计消息中间件的设计思路一般都是基于主题订阅与发布的机制，RocketMQ也不例外。RocketMQ中，消息生产
- 主要是应对这种需求：软件只允许启动一次。将这个问题转化一下，可以这样描述：对于一个软件，在启动一个进程之后，不允许启动其它进程，如果第二次打
- /** * 进行BigDecimal对象的加减乘除，四舍五入等运算的工具类 * * @author Marydon * @createTi
- 闭锁相当于一扇门，在闭锁到达结束状态之前，这扇门一直是关闭着的，没有任何线程可以通过，当到达结束状态时，这扇门才会打开并容许所有线程通过。它
- 1、什么是序列化与反序列化？序列化：指把堆内存中的 Java 对象数据，通过某种方式把对象存储到磁盘文件中或者传递给其他网络节点（在网络上传
- 对于一个App的UI而言，在流畅性上的改进目标其实就是降低屏幕绘制的延迟，创建流畅和稳定的帧率以避免卡顿。在理想情况下，全部的测量、布局和绘
- Android小白第一次写博客，心情无比激动。下面给大家展示一下卫星菜单的实现。1.简单介绍卫星菜单在应用程序中，有很多展示菜单的方式，但其
- 一.hutool工具摘抄一段hutool工具的简介:Hutool是一个小而全的Java工具类库，通过静态方法封装，降低相关API的学习成本，
- 1 前言在前文中，已经讲述了 AOP 的后置处理器使用和方法，在本文中继续分享增强信息相关的源码，这里才是 AOP 的核心代码。2 spri
- 1.使用ASCII码判断您可以使用ASCII码来进行判断字符串中的内容是否为纯数字。步骤如下：先判断字符串是否为空的情况，保证代码运行的稳定
- springboot和springmvc的区别spring boot 内嵌tomcat，Jetty和Undertow容器，可以直接运行起来，
