博客

 

一，RabbitMQ简介

　　在介绍RabbitMQ之前，先让我们共同来了解一下什么是”消息队列”。

       度娘曰：“消息”是在两台计算机间传送的数据单位。消息可以非常简单，例如只包含文本字符串；也可以更复杂，可能包含
嵌入对象。

　　消息被发送到队列中。“消息队列”是在消息的传输过程中保存消息的容器。消息队列管理器在将消息从它的源中继到它的目标时充当中间人。队列的主要目的是提供路由并保证消息的传递；如果发送消息时接收者不可用，消息队列会保留消息，直到可以成功地传递它。

　　而RabbitMQ就是这种保存消息的容器的其中一种，当然还有其他容器：Active MQ,Zero MQ,Kafka等。

　　RabbitMQ是一个开源，跨语言，跨平台的容器，官网
http://www.rabbitmq.com

二，RabbitMQ下载安装

　　RabbitMQ支持多语言跨平台，不同语言可以下载不同的版本。我以C#版本为例将其下载并安装在Windows平台。

　　由于RabbitMQ服务器是基于erlang语言编写的，所以我们需要安装支持Windows 版的Erlang，下载地址：http://www.erlang.org/downloads，选择下图只向文件下载并安装

　　（注：如没有安装Windows 版的Erlang，你就直接下载安装RabbitMQ进行安装，会提示先安装Erlang，在安装RabbitMQ服务器）

下载完成后点击exe进行安装

 

 

 下载安装RabbitMQ服务器：http://www.rabbitmq.com/install-windows-manual.html

 

 

 

同步Erlang Cookie：Erlang cookie是用于RabbitMQ集群节点和CLI工具之间验证的共享密钥。该值存储在通常称为Erlang Cookie文件的文件中。

用户erlang.cookie位置：C:\Users\用户

Windows服务目录位置：C:\Windows\System32\config\systemprofile

位置：要确保Erlang Cookie文件包含相同的字符串，请复制上面列出的Windows服务目录中的.erlang.cookie文件以替换用户.erlang.cookie。

 如何Erlang 设置环境变量请安官网位置查找

 

启用RabbitMQ管理插件：

开始–>RabbitMQ SERVER–>RabbitMQ Commond控制台程序：

执行命令：rabbitmq-plugins enable rabbitmq_management

在浏览器中输入http://localhost:15672/打开管理登录界面

默认用户名密码均为：”guest”

 

在vs中我们可以通过nuget安装RabbitMQ客户端

 

三，RabbitMQ使用

几个重要概念：

Broker：简单来说就是消息队列服务器实体。

　　Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。

　　Queue：消息队列载体，每个消息都会被投入到一个或多个队列。

　　Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。

　　Routing Key：路由关键字，exchange根据这个关键字进行消息投递。

　　vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。

　　producer：消息生产者，就是投递消息的程序。

　　consumer：消息消费者，就是接受消息的程序。

　　channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

消息队列的使用过程，如下：

（1）客户端连接到消息队列服务器，打开一个channel。

（2）客户端声明一个exchange，并设置相关属性。

（3）客户端声明一个queue，并设置相关属性。

（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。

（5）客户端投递消息到exchange。

exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。

 

案例：建立两个控制台程序  发送消息Send.cs

public static void Send(string msg)
        {
　　　　　　//连接字符创
            var factory = new ConnectionFactory() { HostName = "192.168.13.226", VirtualHost = "rabbittest", UserName = "admin",Password = "1234567"};
            
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "logs", type: "fanout");
                var body = Encoding.UTF8.GetBytes(msg);

                //消息确认
                var properties = channel.CreateBasicProperties();
                properties.Persistent = true;

　　　　　　　　　　//发送消息
                channel.BasicPublish(exchange: "logs", routingKey: "", basicProperties: properties, body: body);

            }
        }

 

接收消息：

var factory = new ConnectionFactory() { HostName = "localhost", VirtualHost = "rabbittest" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "logs", type: "fanout");
                //交换器与队列绑定
                   var queueName = channel.QueueDeclare().QueueName;
                channel.QueueBind(queue: queueName,
                    exchange: "logs",
                    routingKey: "");

                Console.WriteLine(" [*] Waiting for logs.");

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] {0}", message);
                };
                channel.BasicConsume(queue: queueName,
                    autoAck: true,
                    consumer: consumer);

                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }

 

 

测试结果：

 

 好了，就到这里吧