RabbitMq VS中简单使用
Posted forever2015
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMq VS中简单使用相关的知识，希望对你有一定的参考价值。
1. 搭建环境
1.1 安装Erlang语言运行环境
由于RabbitMQ使用Erlang语言编写，所以先安装Erlang语言运行环境。
1.2 Erlang([\'ə:læŋ])是一种通用的面向并发的编程语言，它由瑞典电信设备制造商爱立信所辖的CS-Lab开发，目的是创造一种可以应对大规模并发活动的编程语言和运行环境。
使用Erlang来编写分布式应用要简单的多，因为它的分布式机制是透明的：对于程序来说并不知道自己是在分布式运行。Erlang运行时环境是一个虚拟机，有点像Java虚拟机，这样代码一经编 译， 同样可以随处运行。它的运行时系统甚至允许代码在不被中断 的情况下更新。另外如果需要更高效的话，字节代码也可以编译成本地代码运行。
1.3
下载地址：http://www.erlang.org/downloads
（语言环境最好下载最新的，不然下载的RabbitMq服务端可能不支持）
1.4.设置环境变量
手动编辑“path”加入路径 C:\\Program Files\\erl8.2\\bin
1.5.检查Erlang是否安装成功
打开 cmd ,输入 erl 后回车，如果看到如下的信息，表明安装成功。
1.6 安装RabbitMQ服务端
地址 http://www.rabbitmq.com/
使RabbitMQ以Windows Service的方式在后台运行：打开cmd切换到sbin目录下执行（必须一管理员身份运行）
rabbitmq-service install rabbitmq-service enable rabbitmq-service start
现在RabbitMQ的服务端已经启动起来了。
要查看和控制RabbitMQ服务端的状态，可以用rabbitmqctl这个脚本。
比如查看状态：
rabbitmqctl status
假如显示node没有连接上，需要到C:\\Windows目录下，将.erlang.cookie文件，拷贝到用户目录下 C:\\Windows\\System32\\config\\systemprofile 替换其中的.erlang.cookie文件，这是Erlang的Cookie文件，允许与Erlang进行交互。
使用命令查看用户：
rabbitmqctl list_users
RabbitMQ会为我们创建默认的用户名guest和密码guest，guest默认拥有RabbitMQ的所有权限。
一般的，我们需要新建一个我们自己的用户，设置密码，并授予权限，并将其设置为管理员，可以使用下面的命令来执行这一操作：
rabbitmqctl add_user JC JayChou //创建用户JC密码为JayChou rabbitmqctl set_permissions JC ".*" ".*" ".*" //赋予JC读写所有消息队列的权限 rabbitmqctl set_user_tags JC administrator //分配用户组
修改JC密码为123：
rabbitmqctl change_password JC 123
删除用户JC：
rabbitmqctl delete_user JC
也可以开启rabbitmq_management插件，在web界面查看和管理RabbitMQ服务
rabbitmq-plugins enable rabbitmq_management
2.使用（中文网站http://rabbitmq.mr-ping.com/）
Hello Word
下面来展示简单的RabbitMQ的使用：
首先创建控制台项目,使用nuget包添加RabbitMQ.Client.dll。这个程序作为Producer生产者，用来发送数据：
static void Main(string[] args)
{
var factory = new ConnectionFactory();
factory.HostName = "localhost";//RabbitMQ服务在本地运行
factory.UserName = "guest";//用户名
factory.Password = "guest";//密码
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare("hello", false, false, false, null);//创建一个名称为hello的消息队列
string message = "Hello World"; //传递的消息内容
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "hello", null, body); //开始传递
Console.WriteLine("已发送： {0}", message);
Console.ReadLine();
}
}
}
首先，需要创建一个ConnectionFactory，设置目标，由于是在本机，所以设置为localhost，如果RabbitMQ不在本机，只需要设置目标机器的IP地址或者机器名称即可，然后设置前面创建的用户名和密码。
紧接着要创建一个Channel，如果要发送消息，需要创建一个队列，然后将消息发布到这个队列中。在创建队列的时候，只有RabbitMQ上该队列不存在，才会去创建。消息是以二进制数组的形式传输的，所以如果消息是实体对象的话，需要序列化和然后转化为二进制数组。
现在客户端发送代码已经写好了，运行之后，消息会发布到RabbitMQ的消息队列中，现在需要编写服务端的代码连接到RabbitMQ上去获取这些消息。
3.2.2创建的控制台项目，引用RabbitMQ.Client.dll。作为Consumer消费者，用来接收数据：
static void Main(string[] args)
{
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.UserName = "guest";
factory.Password = "guest";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare("hello", false, false, false, null);
var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume("hello", false, consumer);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine("已接收： {0}", message);
};
Console.ReadLine();
}
}
}
和发送一样，首先需要定义连接，然后声明消息队列。要接收消息，需要定义一个Consume，然后在接收消息的事件中处理数据。
发送消息：
现在，名为hello的消息队列中，发送了一条消息。这条消息存储到了RabbitMQ的服务器上了。使用rabbitmqctl 的list_queues可以查看所有的消息队列，以及里面的消息个数，可以看到，目前Rabbitmq上只有一个消息队列，里面只有一条消息：
也可以在web管理界面查看此queue的相关信息：
接收消息：
既然消息已经被接收了，那我们再来看queue的内容:
前面的例子展示了如何在指定的消息队列发送和接收消息。
现在我们创建一个工作队列（work queue）来将一些耗时的任务分发给多个工作者（workers）：
工作队列（work queues, 又称任务队列Task Queues）的主要思想是为了避免立即执行并等待一些占用大量资源、时间的操作完成。而是把任务（Task）当作消息发送到队列中，稍后处理。一个运行在后台的工作者（worker）进程就会取出任务然后处理。当运行多个工作者（workers）时，任务会在它们之间共享。
这个在网络应用中非常有用，它可以在短暂的HTTP请求中处理一些复杂的任务。在一些实时性要求不太高的地方，我们可以处理完主要操作之后，以消息的方式来处理其他的不紧要的操作，比如写日志等等。
准备
在第一部分，发送了一个包含“Hello World!”的字符串消息。现在发送一些字符串，把这些字符串当作复杂的任务。这里使用time.sleep()函数来模拟耗时的任务。在字符串中加上点号（.）来表示任务的复杂程度，一个点（.）将会耗时1秒钟。比如"Hello..."就会耗时3秒钟。
对之前示例的send.cs做些简单的调整，以便可以发送随意的消息。这个程序会按照计划发送任务到我们的工作队列中。
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.UserName = "guest";
factory.Password = "guest";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare("hello", false, false, false, null);
string message = GetMessage(args);
var properties = channel.CreateBasicProperties();
//消息是持久的，存在并不会受服务器重启影响
properties.DeliveryMode = 2;
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "hello", properties, body);
Console.WriteLine(" set {0}", message);
}
}
Console.ReadKey();
}
private static string GetMessage(string[] args)
{
return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
}
接着我们修改接收端，让他根据消息中的逗点的个数来Sleep对应的秒数：
static void Main(string[] args)
{
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.UserName = "yy";
factory.Password = "hello!";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare("hello", false, false, false, null);
var consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume("hello", true, consumer);
while (true)
{
var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
int dots = message.Split(\'.\').Length - 1;
Thread.Sleep(dots * 1000);
Console.WriteLine("Received {0}", message);
Console.WriteLine("Done");
}
}
}
}
轮询分发
使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务，我们只需要添加更多的工作者（workers）就可以了，扩展很简单。
现在，我们先启动两个接收端，等待接受消息，然后启动一个发送端开始发送消息。
在cmd条件下，发送了5条消息，每条消息后面的逗点表示该消息需要执行的时长，来模拟耗时的操作。
然后可以看到，两个接收端依次接收到了发出的消息：
默认，RabbitMQ会将每个消息按照顺序依次分发给下一个消费者。所以每个消费者接收到的消息个数大致是平均的。 这种消息分发的方式称之为轮询（round-robin）。
3.4 消息响应
当处理一个比较耗时得任务的时候，也许想知道消费者（consumers）是否运行到一半就挂掉。在当前的代码中，当RabbitMQ将消息发送给消费者（consumers）之后，马上就会将该消息从队列中移除。此时，如果把处理这个消息的工作者（worker）停掉，正在处理的这条消息就会丢失。同时，所有发送到这个工作者的还没有处理的消息都会丢失。
我们不想丢失任何任务消息。如果一个工作者（worker）挂掉了，我们希望该消息会重新发送给其他的工作者（worker）。
为了防止消息丢失，RabbitMQ提供了消息响应（acknowledgments）机制。消费者会通过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，然后RabbitMQ才会释放并删除这条消息。
如果消费者（consumer）挂掉了，没有发送响应，RabbitMQ就会认为消息没有被完全处理，然后重新发送给其他消费者（consumer）。这样，即使工作者（workers）偶尔的挂掉，也不会丢失消息。
消息是没有超时这个概念的；当工作者与它断开连的时候，RabbitMQ会重新发送消息。这样在处理一个耗时非常长的消息任务的时候就不会出问题了。
消息响应默认是开启的。在之前的例子中使用了no_ack=True标识把它关闭。是时候移除这个标识了，当工作者（worker）完成了任务，就发送一个响应。
channel.BasicConsume("hello", false, consumer);
while (true)
{
var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
int dots = message.Split(\'.\').Length - 1;
Thread.Sleep(dots * 1000);
Console.WriteLine("Received {0}", message);
Console.WriteLine("Done");
channel.BasicAck(ea.DeliveryTag, false);
}
现在,可以保证,即使正在处理消息的工作者被停掉,这些消息也不会丢失,所有没有被应答的消息会被重新发送给其他工作者.
一个很常见的错误就是忘掉了BasicAck这个方法,这个错误很常见,但是后果很严重. 当客户端退出时,待处理的消息就会被重新分发,但是RabitMQ会消耗越来越多的内存,因为这些没有被应答的消息不能够被释放。调试这种case，可以使用rabbitmqct打印messages_unacknoledged字段。
rabbitmqctl list_queues name messages_ready messages_unacknowledged Listing queues ... hello 0 0 ...done.
3.5 消息持久化
前面已经搞定了即使消费者down掉，任务也不会丢失，但是，如果RabbitMQ Server停掉了，那么这些消息还是会丢失。
当RabbitMQ Server 关闭或者崩溃，那么里面存储的队列和消息默认是不会保存下来的。如果要让RabbitMQ保存住消息，需要在两个地方同时设置：需要保证队列和消息都是持久化的。
首先，要保证RabbitMQ不会丢失队列，所以要做如下设置：
bool durable = true;
channel.QueueDeclare("hello", durable, false, false, null);
虽然在语法上是正确的，但是在目前阶段是不正确的，因为我们之前已经定义了一个非持久化的hello队列。RabbitMQ不允许我们使用不同的参数重新定义一个已经存在的同名队列，如果这样做就会报错。现在，定义另外一个不同名称的队列：
bool durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
queueDeclare 这个改动需要在发送端和接收端同时设置。
现在保证了task_queue这个消息队列即使在RabbitMQ Server重启之后，队列也不会丢失。 然后需要保证消息也是持久化的， 这可以通过设置IBasicProperties.SetPersistent 为true来实现：
var properties = channel.CreateBasicProperties(); properties.SetPersistent(true);
需要注意的是，将消息设置为持久化并不能完全保证消息不丢失。虽然他告诉RabbitMQ将消息保存到磁盘上，但是在RabbitMQ接收到消息和将其保存到磁盘上这之间仍然有一个小的时间窗口。 RabbitMQ 可能只是将消息保存到了缓存中，并没有将其写入到磁盘上。持久化是不能够一定保证的，但是对于一个简单任务队列来说已经足够。如果需要消息队列持久化的强保证，可以使用publisher confirms
3.6 公平分发
你可能会注意到，消息的分发可能并没有如我们想要的那样公平分配。比如，对于两个工作者。当奇数个消息的任务比较重，但是偶数个消息任务比较轻时，奇数个工作者始终处理忙碌状态，而偶数个工作者始终处理空闲状态。但是RabbitMQ并不知道这些，他仍然会平均依次的分发消息。
为了改变这一状态，我们可以使用basicQos方法，设置perfetchCount=1 。这样就告诉RabbitMQ 不要在同一时间给一个工作者发送多于1个的消息，或者换句话说。在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
channel.BasicQos(0, 1, false);
3.7 完整实例
现在将所有这些放在一起：
发送端代码如下：
static void Main(string[] args)
{
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.UserName = "yy";
factory.Password = "hello!";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
bool durable = true;
channel.QueueDeclare("task_queue", durable, false, false, null);
string message = GetMessage(args);
var properties = channel.CreateBasicProperties();
properties.SetPersistent(true);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "task_queue", properties, body);
Console.WriteLine(" set {0}", message);
}
}
Console.ReadKey();
}
private static string GetMessage(string[] args)
{
return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
}
接收端代码如下：
static void Main(string[] args)
{
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.UserName = "yy";
factory.Password = "hello!";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
bool durable = true;
channel.QueueDeclare("task_queue", durable, false, false, null);
channel.BasicQos(0, 1, false);
var consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume("task_queue", false, consumer);
while (true)
{
var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
int dots = message.Split(\'.\').Length - 1;
Thread.Sleep(dots * 1000);
Console.WriteLine("Received {0}", message);
Console.WriteLine("Done");
channel.BasicAck(ea.DeliveryTag, false);
}
}
}
}
4 管理界面
RabbitMQ管理界面，通过该界面可以查看RabbitMQ Server 当前的状态，该界面是以插件形式提供的，并且在安装RabbitMQ的时候已经自带了该插件。需要做的是在RabbitMQ控制台界面中启用该插件，命令如下：
rabbitmq-plugins enable rabbitmq_management
现在，在浏览器中输入 http://server-name:15672/ server-name换成机器地址或者域名，如果是本地的，直接用localhost（RabbitMQ 3.0之前版本端口号为55672）在输入之后，弹出登录界面，使用我们之前创建的用户登录。
.
以上是关于RabbitMq VS中简单使用的主要内容，如果未能解决你的问题，请参考以下文章