.NET中RabbitMQ的使用

概述

　　MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。AMQP(高级消息队列协议) 是一个异步消息传递所使用的应用层协议规范，作为线路层协议，而不是API（例如JMS），AMQP 客户端能够无视消息的来源任意发送和接受信息。AMQP的原始用途只是为金融界提供一个可以彼此协作的消息协议，而现在的目标则是为通用消息队列架构提供通用构建工具。因此，面向消息的中间件 （MOM）系统，例如发布/订阅队列，没有作为基本元素实现。AMQP当中有四个概念非常重要（一个虚拟主机持有一组交换机、队列和绑定）：

  1. virtual host，虚拟主机
  2. exchange，交换机
  3. queue，队列
  4. binding，绑定

　　更多理论性东西可以参考（在Windows上安装Rabbit MQ 指南），针对队列的讲解相当详细

Window下安装RabbbitMQ

文件下载安装

Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装Rabbit MQ的前提是安装Erlang。通过下面两个连接下载安装3.2.3 版本：

  1. 下载并安装 Erlang OTP For Windows (vR16B03)
  2. 运行安装 Rabbit MQ Server Windows Installer (v3.2.3)

默认安装的Rabbit MQ 监听端口是5672。先安装Erlang OTP后安装RabbitMQ，安装方式默认即可，RabbitMQ可以勾选安装后台服务、服务启动和停止等操作。

激活Rabbit MQ's Management Plugin

使用Rabbit MQ 管理插件，可以更好的可视化方式查看Rabbit MQ 服务器实例的状态，打开CMD命令，cd到安装目录（..\rabbitmq_server-3.2.3\sbin）下，输入下面的命令激活：

rabbitmq-plugins enable rabbitmq_management

要重启服务才能生效，可以执行

net stop RabbitMQ && net start RabbitMQ

输入网址，打开监控页面：  http://localhost:15672 (默认账号和密码：guest 和guest)

配置RabbitMQ用户权限

RabbitMQ是存在用户权限的，默认是guest 密码也是guest，隶属于Administrator管理员下。现需要配置新用户和权限，继续打开CMD命令，cd到安装目录sbin下：

用户操作指令：

复制代码
::查询服务状态
rabbitmqctl status

::列举虚拟主机列表
rabbitmqctl list_vhosts
::列举用户列表
rabbitmqctl list_users

:: 添加用户和密码
rabbitmqctl  add_user  hao  abc123:: 设置权限   
rabbitmqctl  set_permissions  yy  ".*"  ".*"  ".*":: 分配用户组
rabbitmqctl  set_user_tags yy administrator
:: 删除guest用户
rabbitmqctl delete_user guest
::修改用户密码
rabbitmqctl change_password {username}  {newpassowrd}
复制代码

 

.NET中RabbitMQ使用

1、Nuget下载RabbitMQ.Client第三方类库，版本V3.6.5，高版本与.NET Framework 4.5有冲突，RabbitMQ Client地址

2、利用RabbitMQ Clinet类库编码(代码内容有注释，此处不做详细解释，文章后有完整代码的下载地址)

 

　　<1>RabbitMQ的direct类型Exchange

　　　Producter发送消息代码：

复制代码
        /// <summary>/// 连接配置/// </summary>private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){HostName ="192.168.1.8",UserName="hao",Password="abc123",Port= 5672};/// <summary>/// 路由名称/// </summary>const string ExchangeName = "justin.exchange";//队列名称const string QueueName = "justin.queue";public static void DirectExchangeSendMsg(){using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);var props = channel.CreateBasicProperties();props.Persistent = true;string vadata = Console.ReadLine();while (vadata != "exit"){var msgBody = Encoding.UTF8.GetBytes(vadata);channel.BasicPublish(exchange: ExchangeName, routingKey: QueueName, basicProperties: props, body: msgBody);        Console.WriteLine(string.Format("***发送时间:{0}，发送完成，输入exit退出消息发送",DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));vadata = Console.ReadLine();}                   }}}
复制代码

　　　

　　Customer接收消息代码：

 

复制代码
        /// <summary>/// 连接配置/// </summary>private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {HostName = "192.168.1.8", UserName = "hao", Password = "abc123", Port = 5672};/// <summary>/// 路由名称/// </summary>const string ExchangeName = "justin.exchange";//队列名称const string QueueName = "justin.queue";public static void DirectAcceptExchange(){using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);while (true){ BasicGetResult msgResponse = channel.BasicGet(QueueName, noAck: true);if (msgResponse != null){var msgBody = Encoding.UTF8.GetString(msgResponse.Body);Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"),msgBody));}//BasicGetResult msgResponse2 = channel.BasicGet(QueueName, noAck: false);process message ...//channel.BasicAck(msgResponse2.DeliveryTag, multiple: false);System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));}}}}
复制代码

　　

　　但是这种处理速度较慢，因为循环线程等待。高效的接收消息的方式可以使用EventingBasicConsumer进行消息接收处理，修改代码内容如下：

 

复制代码
        public static void DirectAcceptExchangeEvent(){using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){//channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);//channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);var consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{                        var msgBody = Encoding.UTF8.GetString(ea.Body);Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));};channel.BasicConsume(QueueName, noAck: true, consumer: consumer);//已过时用EventingBasicConsumer代替//var consumer2 = new QueueingBasicConsumer(channel);//channel.BasicConsume(QueueName, noAck: true, consumer: consumer);//var msgResponse = consumer2.Queue.Dequeue(); //blocking//var msgBody2 = Encoding.UTF8.GetString(msgResponse.Body);Console.WriteLine("按任意值，退出程序");Console.ReadKey();}}}
复制代码

　　

　　但是有些时候，消费者同一时间没有能力处理太多的业务，导致分配过来的队列消息不能及时处理完成，这个时候，我们可以设置BasicQos属性，告诉Broker同一时间将未处理完成的消息分配其他消费者，所以接收消息的地方需要略做修改，代码如下：

 

复制代码
public static void DirectAcceptExchangeTask()
{using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){//channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//告诉broker同一时间只处理一个消息//channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);var consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{var msgBody = Encoding.UTF8.GetString(ea.Body);Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));int dots = msgBody.Split('.').Length - 1;System.Threading.Thread.Sleep(dots * 1000);Console.WriteLine(" [x] Done");//处理完成，告诉Broker可以服务端可以删除消息，分配新的消息过来channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);};//noAck设置false,告诉broker，发送消息之后，消息暂时不要删除，等消费者处理完成再说channel.BasicConsume(QueueName, noAck: false, consumer: consumer);Console.WriteLine("按任意值，退出程序");Console.ReadKey();}}
}
复制代码

　　

　　<2> RabbitMQ的Topic类型Exchange

　　Producter 发送消息代码：

复制代码
        /// <summary>/// 连接配置/// </summary>private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){HostName ="192.168.1.8",UserName="hao",Password="abc123",Port= 5672};/// <summary>/// 路由名称/// </summary>const string TopExchangeName = "topic.justin.exchange";//队列名称const string TopQueueName = "topic.justin.queue";public static void TopicExchangeSendMsg(){using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);//var props = channel.CreateBasicProperties();//props.Persistent = true;string vadata = Console.ReadLine();while (vadata != "exit"){var msgBody = Encoding.UTF8.GetBytes(vadata);channel.BasicPublish(exchange: TopExchangeName, routingKey: TopQueueName, basicProperties: null, body: msgBody);Console.WriteLine(string.Format("***发送时间:{0}，发送完成，输入exit退出消息发送", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));vadata = Console.ReadLine();}}}}
复制代码

 

　　Customer接收消息代码：

 

复制代码
        /// <summary>/// 连接配置/// </summary>private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {HostName = "192.168.1.8", UserName = "hao", Password = "abc123", Port = 5672};/// <summary>/// 路由名称/// </summary>const string TopExchangeName = "topic.justin.exchange";//队列名称const string TopQueueName = "topic.justin.queue";public static void TopicAcceptExchange(){using (IConnection conn = rabbitMqFactory.CreateConnection()){using (IModel channel = conn.CreateModel()){channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);var consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{var msgBody = Encoding.UTF8.GetString(ea.Body);Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));int dots = msgBody.Split('.').Length - 1;System.Threading.Thread.Sleep(dots * 1000);Console.WriteLine(" [x] Done");channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);};channel.BasicConsume(TopQueueName, noAck: false, consumer: consumer);Console.WriteLine("按任意值，退出程序");Console.ReadKey();}}}
复制代码

参考资料：

在 Windows 上安装Rabbit MQ 指南（http://www.cnblogs.com/shanyou/p/4067250.html）

.NET 环境中使用RabbitMQ（http://www.cnblogs.com/yangecnu/p/4227535.html）

RabbitMQ Tutorial（http://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html）

源代码下载

知道的越多，不知道的也就越多，多多学习！

转载于:https://www.cnblogs.com/Jeely/p/10788484.html

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

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

相关文章

SQL Server死锁诊断--同一行数据在不同索引操作下引起的死锁

SQL Server死锁诊断--同一行数据在不同索引操作下引起的死锁

死锁概述 对于数据库中出现的死锁&#xff0c;通俗地解释就是&#xff1a;不同Session&#xff08;会话&#xff09;持有一部分资源&#xff0c;并且同时相互排他性地申请对方持有的资源&#xff0c;然后双方都得不到自己想要的资源&#xff0c;从而造成的一种僵持的现象。当然…
阅读更多...
python下载安装搭建

python下载安装搭建

python官网下载python运行环境&#xff08;https://www.python.org/downloads/&#xff09;&#xff0c;建议下载稳定版本&#xff0c;不推荐使用最新版本 安装 然后我们打开CMD&#xff0c;在里面输入python&#xff0c;就可以直接进入进行编码了 如果输入python出现下面错误 …
阅读更多...
35个Java 代码性能优化总结

35个Java 代码性能优化总结

前言代码优化&#xff0c;一个很重要的课题。可能有些人觉得没用&#xff0c;一些细小的地方有什么好修改的&#xff0c;改与不改对于代码的运行效率有什么影响呢&#xff1f;这个问题我是这么考虑的&#xff0c;就像大海里面的鲸鱼一样&#xff0c;它吃一条小虾米有用吗&#…
阅读更多...
MySQL讲义

MySQL讲义

1 MySQL基础知识 瑞典MySQL AB公司开发&#xff0c;由SUN收购&#xff0c;而后SUN被甲骨文并购&#xff0c;目前属于Oracle公司。 MySQL是一种关联数据库管理系统 由于其体积小、速度快、总体拥有成本低、MySQL软件采用了双授权政策&#xff0c;分为社区版和企业版。 …
阅读更多...
Teams Bot App Manifest 文件解析

Teams Bot App Manifest 文件解析

这篇文章我们继续以 Hello World Bot 这个 sample 来讲一下 manifest template。 实际上在 Teams app 开发的时候&#xff0c;有 manifest 的概念&#xff0c;manifest 是用来说明这个 teams app 的一些基本信息和配置信息&#xff0c;比如 app 的名字&#xff0c;app有哪些能…
阅读更多...
[Dart] Flutter开发中的几个常用函数

[Dart] Flutter开发中的几个常用函数

几个Flutter开发中的常用函数 /** 返回当前时间戳 */static int currentTimeMillis() {return new DateTime.now().millisecondsSinceEpoch;}/** 复制到剪粘板 */static copyToClipboard(final String text) {if (text null) return;Clipboard.setData(new ClipboardData(text…
阅读更多...
Cordova入门系列（三）Cordova插件调用 转发 https://www.cnblogs.com/lishuxue/p/6018416.html...

Cordova入门系列（三）Cordova插件调用 转发 https://www.cnblogs.com/lishuxue/p/6018416.html...

Cordova入门系列&#xff08;三&#xff09;Cordova插件调用 版权声明&#xff1a;本文为博主原创文章&#xff0c;转载请注明出处 上一章我们介绍了cordova android项目是如何运行的&#xff0c;这一章我们介绍cordova的核心内容&#xff0c;插件的调用。演示一个例子&#xf…
阅读更多...
clojure with postgres

clojure with postgres

主要关注访问pg。不关心其他db 1 clojure.java.jdbc https://github.com/clojure/java.jdbchttp://clojure-doc.org/articles/ecosystem/java_jdbc/reusing_connections.html这个最广&#xff0c;需要配合不同DB[org.clojure/java.jdbc "0.7.9"] [org.postgresql/pos…
阅读更多...
lua入门

lua入门

https://en.blog.nic.cz/2015/08/12/embedding-luajit-in-30-minutes-or-so/
阅读更多...
shell脚本传可选参数 getopts 和 getopt的方法

shell脚本传可选参数 getopts 和 getopt的方法

写了一个shell脚本&#xff0c;需要向shell脚本中传参数供脚本使用&#xff0c;达到的效果是传的参数可以是可选参数 下面是一个常规化的shell脚本&#xff1a; echo "执行的文件名为: $0";echo "第一个参数名为: $1";echo "第二个参数名为: $2"…
阅读更多...
Teams Tab App 代码深入浅出 - 配置页面

Teams Tab App 代码深入浅出 - 配置页面

上一篇文章我们使用Teams Toolkit 来创建、运行 tab app。这篇文章我们深入来分析看一下tab app 的代码。 先打开代码目录&#xff0c;可以看到在 src 目录下有入口文件 index.tsx&#xff0c;然后在 components 目录下有更多的一些 tsx 文件&#xff0c;tsx 是 typescript的一…
阅读更多...
labelme标注的json文件数据转成coco数据集格式（可处理目标框和实例分割）

labelme标注的json文件数据转成coco数据集格式（可处理目标框和实例分割）

这里主要是搬运一下能找到的 labelme标注的json文件数据转成coco数据集格式&#xff08;可处理目标框和实例分割&#xff09;的代码&#xff0c;以供需要时参考和提供相关帮助。 1、官方labelme实现 如下是labelme官方网址&#xff0c;提供了源代码&#xff0c;以及相关使用方…
阅读更多...
EpSON TM-82II驱动在POS系统上面安装问题处理

EpSON TM-82II驱动在POS系统上面安装问题处理

按照品牌名称&#xff0c;在网上下载的安装包为apstmt82.rar 下面讲解一下&#xff0c;如何的解决爱普生打印机在POS机器上面的安装问题&#xff0c;这个算是一个比较奇特的故障问题&#xff0c;不像其它的新北冰洋&#xff08;SN3C&#xff09;的U80_U80II&#xff0c;SeNor的…
阅读更多...
打印图片的属性和实现另存图片功能以及使用numpy

打印图片的属性和实现另存图片功能以及使用numpy

上一篇我们已经学了如何读取图片的功能了以及和opencv的环境搭建了&#xff0c;今天接着来学习&#xff0c;哈哈哈&#xff0c;今天刚好五一&#xff0c;也没闲着&#xff0c;继续学习。 1、 首先我们来实现打印出图片的一些属性功能&#xff0c; 先来看一段代码&#xff1a; 1…
阅读更多...
Ubuntu 18.04下命令安装VMware Tools

Ubuntu 18.04下命令安装VMware Tools

2019独角兽企业重金招聘Python工程师标准>>> sudo apt-get upgrade sudo apt-get install open-vm-tools-desktop -y sudo reboot 转载于:https://my.oschina.net/u/574036/blog/1829455
阅读更多...
phpstorm PHP language level无法选择

phpstorm PHP language level无法选择

phpstorm PHP7新特性一直提示红色波浪线&#xff0c;应该是没有设置PHP 版本&#xff0c;但是打开PHPstorm---preference--lannguage&frameworks--PHP &#xff0c; 发现PHP language level 无法选择PHP7.2 &#xff0c;查看旁边的提示信息说是同步了composer 的原因&#…
阅读更多...
Qfile

Qfile

打开方式&#xff1a; 1 void AddStudents::write_to_file(QString src){2 QFile file("stu.txt");3 if (!file.open(QIODevice::Append | QIODevice::Text)){4 QMessageBox::critical(this,"打开文件错误","确认");5 r…
阅读更多...
多层装饰器、带参数装饰器

多层装饰器、带参数装饰器

# 带参数的装饰器 # import time # FLAGE False # 加个标志位&#xff0c;使全部的装饰器可以失效或有效 # def timmer_out(flag): # def timmer(func): # def inner(*args,**kwargs): # if flag: # start time.time() # …
阅读更多...
IDEA svn 菜单不见了,解决方法

IDEA svn 菜单不见了,解决方法

2019独角兽企业重金招聘Python工程师标准>>> 参考地址: http://www.cnblogs.com/signheart/p/193448a98f92bd0cc064dbd772dd9f48.html,我是第二种方法解决的! 转载于:https://my.oschina.net/liuchangng/blog/1829679
阅读更多...
苏宁易购：Hadoop失宠前提是出现更强替代品

苏宁易购：Hadoop失宠前提是出现更强替代品

在笔者持续调研国内Hadoop生态系统生存现状的同时&#xff0c;KDnuggets发布的2018年数据科学和机器学习工具调查报告再次将“Hadoop失宠”言论复活。报告一出&#xff0c;“Hadoop被抛弃”几个字瞬时成为各大标题党的最爱&#xff0c;充斥在不同的新闻平台。这些报告和数据是否…
阅读更多...
最新文章

Copyright @ 2022~2023