.NET中RabbitMQ的使用

概述

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

virtual host，虚拟主机
exchange，交换机
queue，队列
binding，绑定

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

Window下安装RabbbitMQ

文件下载安装

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

下载并安装 Erlang OTP For Windows (vR16B03)
运行安装 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

.NET中RabbitMQ的使用相关推荐

如何访问Linux服务器中RabbitMQ管理页面
大家好,本篇文章主要讲的是解决访问不到Linux服务器中RabbitMQ管理页面问题,感兴趣的同学赶快来看一看吧,对你有帮助的话记得收藏一下,方便下次浏览由于自己项目的某项功能需要服务器中发送消息到 ...
Python中RabbitMQ的使用
Python中RabbitMQ的使用文章目录 Python中RabbitMQ的使用一.AMQP协议 1.工作过程 2.细节问题 3.Exchange交换机主题交换机: 4.Queue队列 5.C ...
刚安装的python如何使用-python中RabbitMQ的使用（安装和简单教程）
1,简介 RabbitMQ(Rabbit Message Queue)是流行的开源消息队列系统,用erlang语言开发. 1.1关键词说明: Broker:消息队列服务器实体. Exchange:消息 ...
Java笔记-Spring中RabbitMQ的调用
目录基本概念代码与演示基本概念 Spring中已经整合了RabbitMQ,通过配置bean文件,然后在源码中加载,可以简化代码操作. 身为C++程序员不得不说这种方式真的是太爽了. 在本人做过的 ...
mac中rabbitmq的安装
安装brew,在命令窗口执行:/usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/instal ...
详解SpringCloud中RabbitMQ消息队列原理及配置，一篇就够！
作者:kosamino cnblogs.com/jing99/p/11679426.html 一.MQ用途 1.同步变异步消息场景:用户下单完成后,发送邮件和短信通知. 运用消息队列之后,用户下单完 ...
KSO-在NETCore中RabbitMQ的使用以及相关代码
安装与配置下载地址 Erlang https://www.erlang-solutions.com/resources/download.html rabbitMQ http://www.rabbi ...
php中rabbitmq消息乱码,PHP实现RabbitMQ消息队列（转）
本篇文章给大家带来的内容是关于PHP和RabbitMQ实现消息队列的完整代码,有一定的参考价值,有需要的朋友可以参考一下,希望对你有所帮助. 先安装PHP对应的RabbitMQ,这里用的是 php_a ...
windows和linux中RabbitMQ客户端下载安装
RabbitMQ的安装与客户端的简单实用本文主要内容是RabbitMQ的安装步骤[Windows系统与linux上的安装]及客户端的简单使用. Windows上安装 1 安装Erlang 下载erl ...

.NET中RabbitMQ的使用

.NET中RabbitMQ的使用

.NET中RabbitMQ的使用相关推荐

最新文章

热门文章