本文内容整理自https://blog.csdn.net/by_ron/category_6167618.html
RabbitMQ–环境搭建
能点进来相信你明白RabbitMQ是干什么的了，这个系列主要是本人根据RabbitMQ官网C#的Tutorials总结出来，旨在为新手提供手把手教学。接受大牛批评指正，共同学习，共同促进。
一、Erlang安装
RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器，分布式处理能力出众，对程序完全透明。在安装RabbitMQ服务之前必须先安装erlang，否则会发生什么呢？嘿嘿，你安装服务时会遭到提示啦。点击Erlang下载地址，进入下载页面
根据自己的Windows版本选择32位或者64位下载安装即可。安装没什么需要注意的（安装目录默认C盘），下一步到finish即可。
二、安装RabbitMQ Server
RabbitMQ Server下载地址
不用多说，安装点击下一步至finish即可。 实际应用中，肯定是要安装在专门服务器上。本文为演示，安装在自己电脑上指定端口，模拟分布式环境（呃，有点牵强，呵呵！）。 你一定还想知道这个服务是用来干嘛的，问得好。这个服务才是兔子的大脑，我们所有的操作都是通过服务来调度的。简单点理解，他就是一个大队列，我们product产生消息之后由它推送给customer。
三、下载驱动dll
点击dll下载地址进入下载页面
下载后
RabbitMQ.client.dll封装了访问服务端所需的API，重要的事情说三遍， 只支持dotnet-4.5以上的项目！ 只支持dotnet-4.5以上的项目！ 只支持dotnet-4.5以上的项目！
第一篇博文，比较简洁，着急写完，有点像完任务似的。但无论如何，我都相信，事情会越来越好的。 什么？违反墨菲定律，没错，就是这么自信！
生产者–消费者模式
上一篇讨论了如何搭建我们的开发环境，无论使用哪种语言，服务的部署肯定都是相同的。 摘自官网：RabbitMQ is a message broker. In essence, it accepts messages from producers, and delivers them to consumers. In-between, it can route, buffer, and persist the messages according to rules you give it.（译文：兔子就是消息代理，本质上，它接受来自producers的消息然后分发给consumers，在此过程中，它能够根据你定制的规则来路由、缓存和持久化消息）。 本文开始将基于生产者-消费者模式创建第一个项目，hello world，是否感觉很亲切呢！
既然是生产者-消费者模式，那么显然意味着我们会有生产者和消费者两套程序。生产者负责生产message并推送给queue，而消费者从queue中拉取消息进行处理。
生产者
首先我们需要创建一个控制台应用程序，生产者，即消息发送方，我们创建一个类Send.cs，当然，如果你愿意，也可以叫Producer.cs或者F**k.cs等等。 还记否，上一篇我们已经下载好了驱动，即RabbitMQ.Client.dll，现在只要在此项目中引用即可。
代码中这样引用即可（哎，官网就是这么详细，稍有常识的猿直接跳过）
代码语言：javascript复制using System;
using System.Text;
using RabbitMQ.Client;123接下来就可以和RabbitMQ Server创建连接了：
代码语言：javascript复制class Send
{
public static void Main()
{
var factory = new ConnectionFactory() { HostName = "localhost" };//创建代理服务器实例。注意：HostName为Rabbit Server所在服务器的ip或域名，如果服务装在本机，则为localhost，默认端口5672
using (var connection = factory.CreateConnection())//创建socket连接
{
using (var channel = connection.CreateModel())//channel中包含几乎所有的api来供我们操作queue
{
...
}
}
}
}很简单的对不对，接下来给出Send.cs的完整代码，为了方便理解，注释我会写在代码中：
代码语言：javascript复制using System;
using RabbitMQ.Client;
using System.Text;
class Send
{
public static void Main()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using(var connection = factory.CreateConnection())
using(var channel = connection.CreateModel())
{
//声明queue
channel.QueueDeclare(queue: "hello",//队列名
durable: false,//是否持久化
exclusive: false,//true:排他性，该队列仅对首次申明它的连接可见，并在连接断开时自动删除
autoDelete: false,//true:如果该队列没有任何订阅的消费者的话，该队列会被自动删除
arguments: null);//如果安装了队列优先级插件则可以设置优先级
string message = "Hello World!";//待发送的消息
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",//exchange名称
routingKey: "hello",//如果存在exchange,则消息被发送到名称为hello的queue的客户端
basicProperties: null,
body: body);//消息体
Console.WriteLine(" [x] Sent {0}", message);
}
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}好的，接下来启动程序，我们的消息就被推送到了Rabbit Server上的queue中，等待客户端的连接，也就是等待消费者拉取。如下图：
消费者
这次重新创建控制台应用程序，类名为Receive.cs，同理，你可以用自己舒服的单词去命名。
代码语言：javascript复制using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
class Receive
{
public static void Main()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using(var connection = factory.CreateConnection())
using(var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "hello",//指定发送消息的queue，和生产者的queue匹配
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
var consumer = new EventingBasicConsumer(channel);
//注册接收事件，一旦创建连接就去拉取消息
consumer.Received = (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
};
channel.BasicConsume(queue: "hello",
noAck: true,//和tcp协议的ack一样，为false则服务端必须在收到客户端的回执（ack）后才能删除本条消息
consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}这段代码是不是和Send很相似呢，没错，在创建连接和声明queue上没有区别！ 运行程序，效果如下：
哪怕你的Send程序已关闭，但只要运行过且成功发送了，queue就会一直保存消息，直到客户端连接，这些消息才会一股脑儿发送给消费者。 你可以这样实验，在bin中debug目录下启动Send.exe，连续3次，然后再运行客户端，就会收到3条消息，如下图：
至此，我们的Hello World已经成功跑起。这个小demo当然不是仅仅用来say hello的，更多的用意是帮助我理解兔子的基本原理，提供一种高并发情形下的解决方案。相信以后公司商城发展壮大时能够用到！！！
工作队列
使用场景：Work Queue被用以处理大量耗时任务，与其等待任务处理完毕耗费大量cpu资源，还不如立即返回并交由代理worker随后处理message。
消息持久化
生产者和消费者的代码和上一节Publish-Consumer基本相同，唯一不同的是配置项的参数调整。代码如下：
代码语言：javascript复制using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
namespace NewTask
{
class NewTask
{
public static void Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "task_queue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
//At this point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by setting IBasicProperties.SetPersistent to true.
var properties = channel.CreateBasicProperties();
//properties.SetPersistent(true);//此方法已经过时:“这种设置方法已经被摒弃，现在使用持久化属性取代它”
properties.Persistent = true;//取代上面的写法
...
}
}
}队列持久化通过durable: true声明，服务重启后队列依然存在，但如果声明为排他队列exclusive: true,则不受持久化影响，连接断开即移除。 消息持久化通过properties.Persistent = true来设置，前提是队列持久化，否则服务宕掉后消息肯定丢失，因为消息的载体队列都没了。
消息回执
在一些对准确度要求比较高的场景下时，我们可能需要收到从消费者传回的ack后才从队列删除。
- 收不到ack，消息会一直驻留在队列中直到连接断开，此时会发送到集群中下一个消费者去处理；
- 队列永远不断开呢？那么消息当然永驻队列中直到内存吃尽，这可不是好事情，所以消费端切记做好异常处理并且finally发送回执；
- 没有集群则连接重新打开就会再次发送给原来的消费者；
谈回worker
在我看来，worker只是消费者的另外一种叫法，只是它的功能更加特殊，本文开头也说了，消费端基本上声明队列，注册接受事件这些步骤都一样，只是配置项的不同罢了，看下worker的配置：
代码语言：javascript复制using System;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace Worker
{
class Worker
{
public static void Main()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "task_queue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: null);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//prefetchCount:This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy（Fair dispatch（公平分发）：它会选择空闲的worker去处理消息，适用于对性能敏感的场景；Round-robin dispatching（轮循分发）：不解释）
Console.WriteLine(" [*] Waiting for messages.");
var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume(queue: "task_queue",
noAck: false,
consumer: consumer);
consumer.Received = (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
int dots = message.Split('.').Length - 1;
Thread.Sleep(dots * 1000);
Console.WriteLine(" [x] Done");
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);//用以处理完毕后发送ack通知服务端
};
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
}注意这句代码以及注释：channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false) 下面是BasicQops方法的说明：
BasicQops用于设置任务分发，只有收到前一条消息的回执才会发给当前worker，否则轮循下一个worker。如果所有worker都是忙碌，那么建议添加worker设备。
说明worker的特点就是单线程处理消息，如果处于忙碌状态（未发送回执），则不会收到队列推送的消息。
小结：通过队列声明durable: true设置队列持久化，并无多大意义，只是为消息持久化做铺垫；消息持久化通过properties.Persistent = true或者properties.DeliveryMode = 2进行设置；消息回执需要注意处理消费端的代码保证ack总是返回给队列；worker实际上就是一种单线程机制的消费者
