在上篇文章中，我们解决了从发送端（Producer）向接收端（Consumer）发送“Hello World”的问题。在实际的应用场景中，这是远远不够的。从本篇文章开始，我们将结合更加实际的应用场景来讲解更多的高级用法。
当有Consumer需要大量的运算时，RabbitMQ Server需要一定的分发机制来balance每个Consumer的load。接下来我们分布讲解。
应用场景就是RabbitMQ Server会将queue的Message分发给不同的Consumer以处理计算密集型的任务：
1. Message acknowledgment 消息确认
每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中，Consumer出错了，异常退出了，而数据还没有处理完成，那么 非常不幸，这段数据就丢失了。因为我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完 成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。
如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。
为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。
在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。
如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。
这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。
这样即使你通过Ctr-C中断了Recieve.cs，那么Message也不会丢失了，它会被分发到下一个Consumer。
如果忘记了ack，那么后果很严重。当Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于 RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。去调试这种错误，可以通过一下命令打印un-acked Messages.
2. Round-robin dispatching 循环分发
RabbitMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的。如果现在load加重，那么只需要创建更多的Consumer来进行任务处理即 可。当然了，对于负载还要加大怎么办？我没有遇到过这种情况，那就可以创建多个virtual Host，细化不同的通信类别了。
1、首先开启两个Consumer，即运行两个Recieve.cs。
2、在开启两个Producer，即运行两个Producer.cs。
默认情况下，RabbitMQ 会顺序的分发每个Message。当每个收到ack后，会将该Message删除，然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin（优雅分发）。
Producer.cs
1 class Program
2 {
3 static void Main(string[] args)
4 {
5 ConnectionFactory factory = new ConnectionFactory() { HostName = "localhost" };
6 using (IConnection connection = factory.CreateConnection())
7 {
8 using (IModel channel = connection.CreateModel())
9 {
10 channel.QueueDeclare("hello", false, false, false, null);
11 var message = GetMessage(args);
12 var body = Encoding.UTF8.GetBytes(message);
13
14 var properties = channel.CreateBasicProperties();
15 properties.DeliveryMode = 2;//non-persistent (1) or persistent (2)
16 //channel.TxSelect();
17 channel.BasicPublish("", "hello", properties, body);
18 //channel.TxCommit();
19 }
20 }
21 }
22
23 private static string GetMessage(string[] args)
24 {
25 return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
26 }
27 }
Consumer.cs
1 //#define demo1
2 #define demo2
3 using RabbitMQ.Client;
4 using RabbitMQ.Client.Events;
5 using System;
6 using System.Collections.Generic;
7 using System.Linq;
8 using System.Text;
9 using System.Threading;
10 using System.Threading.Tasks;
11
12 namespace ReceiveDemo2
13 {
14 /// <summary>
15 /// 一个Send和多个Receive的例子,
16 /// 还加上了ack的例子.
17 /// 优雅分发
18 /// </summary>
19 class Program
20 {
21 static void Main(string[] args)
22 {
23 var factory = new ConnectionFactory() { HostName = "localhost" };
24 using (var connection = factory.CreateConnection())
25 {
26 using (var channel = connection.CreateModel())
27 {
28 channel.QueueDeclare("hello", false, false, false, null);
29 var consumer = new QueueingBasicConsumer(channel);
30 #if demo1
31 channel.BasicConsume("hello", true, consumer);//自动删除消息
32 #else
33 channel.BasicConsume("hello", false, consumer);//需要接受方发送ack回执,删除消息
34 #endif
35 Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
36 while (true)
37 {
38 var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作
39 #if demo2
40 channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应
41 #endif
42 var body = ea.Body;
43 var message = Encoding.UTF8.GetString(body);
44 Console.WriteLine(" [x] Received {0}", message);
45 int dots = message.Split(‘.‘).Length - 1;
46 Thread.Sleep(dots * 1000);
47 Console.WriteLine(" [x] Done");
48 #if demo2
49 //channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应,这句话写道40行和49行运行结果就会不一样.写到这里会发生如果输出[x] Received {0}之后,没有输出 [x] Done之前,CTRL+C结束程序,那么message会自动发给另外一个客户端,当另外一个客户端收到message后,执行完49行的命令之后,服务器会删掉这个message
50 #endif
51 }
52 }
53 }
54 }
55 }
56 }
3. Message durability消息持久化
在上一节中我们知道了即使Consumer异常退出，Message也不会丢失。但是如果RabbitMQ Server退出呢？软件都有bug，即使RabbitMQ Server是完美毫无bug的（当然这是不可能的，是软件就有bug，没有bug的那不叫软件），它还是有可能退出的：被其它软件影响，或者系统重启 了，系统panic了。。。
为了保证在RabbitMQ退出或者crash了数据仍没有丢失，需要将queue和Message都要持久化。queue的持久化需要在声明时指定durable=True，修改Producer和Consumer的channel.QueueDeclare代码，再次强调，Producer和Consumer都应该去创建这个queue，尽管只有一个地方的创建是真正起作用的：
bool durable = true;
channel.QueueDeclare("hello", durable, false, false, null);
上述语句执行不会有什么错误，但是确得不到我们想要的结果，原因就是RabbitMQ Server已经维护了一个叫hello的queue，那么上述执行不会有任何的作用，也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。
那么workaround也很简单，声明一个另外的名字的queue，比如名字定位task_hello，或者通过监控http://localhost:15672/，删除名为“hello”的Queue。
接下来，还需要持久化Message，即在Producer.cs里面Publish的时候指定一个properties，方式如下：
1 static void Main(string[] args)
2 {
3 var factory = new ConnectionFactory() { HostName = "localhost" };
4 using (var connection = factory.CreateConnection())
5 {
6 using (var channel = connection.CreateModel())
7 {
8 bool durable = true;
9 channel.QueueDeclare("task_queue", durable, false, false, null);//queue的持久化需要在声明时指定durable=True
10 var message = GetMessage(args);
11 var body = Encoding.UTF8.GetBytes(message);
12 var properties = channel.CreateBasicProperties();
13 properties.SetPersistent(true);//需要持久化Message，即在Publish的时候指定一个properties，
14 channel.BasicPublish("", "task_hello", properties, body);
15 }
16 }
17 }
关于持久化的进一步讨论：
为了数据不丢失，我们采用了：
- 在数据处理结束后发送ack，这样RabbitMQ Server会认为Message Deliver 成功。
- 持久化queue，可以防止RabbitMQ Server 重启或者crash引起的数据丢失。
- 持久化Message，理由同上。
但是这样能保证数据100%不丢失吗？
答案是否定的。问题就在与RabbitMQ需要时间去把这些信息存到磁盘上，这个time
window虽然短，但是它的确还是有。在这个时间窗口内如果数据没有保存，数据还会丢失。还有另一个原因就是RabbitMQ并不是为每个Message都做fsync：它可能仅仅是把它保存到Cache里，还没来得及保存到物理磁盘上。
因此这个持久化还是有问题。但是对于大多数应用来说，这已经足够了。当然为了保持一致性，你可以把每次的publish放到一个transaction中。这个transaction的实现需要user defined codes。
那么商业系统会做什么呢？一种可能的方案是在系统panic时或者异常重启时或者断电时，应该给各个应用留出时间去flash cache，保证每个应用都能exit gracefully。
4. Fair dispatch 公平分发
你可能也注意到了，分发机制不是那么优雅。默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message，只是按照这个默认机制进行分发。
那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却是毫无休息的机会。那么，RabbitMQ是如何处理这种问题呢？
通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。 设置方法如下：
channel.BasicQos(0, 1, false);
注意，这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。
Consumer.cs
1 static void Main(string[] args)
2 {
3 var factory = new ConnectionFactory() { HostName = "localhost" };
4 using (var connection = factory.CreateConnection())
5 {
6 using (var channel = connection.CreateModel())
7 {
8 bool durable = true;
9 channel.QueueDeclare("task_queue", durable, false, false, null);
10 channel.BasicQos(0, 1, false);//这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
11 var consumer = new QueueingBasicConsumer(channel);
12 channel.BasicConsume("task_hello", false, null, consumer);//需要接受方发送ack回执,删除消息
13 Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
14 while (true)
15 {
16 var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作
17 channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("task_queue", false, null, consumer);对应
18 var body = ea.Body;
19 var message = Encoding.UTF8.GetString(body);
20 Console.WriteLine(" [x] Received {0}", message);
21 int dots = message.Split(‘.‘).Length - 1;
22 Thread.Sleep(dots * 1000);
23 Console.WriteLine(" [x] Done");
24 }
25 }
26 }
27 }
转：
http://www.rabbitmq.com/tutorials/tutorial-two-dotnet.html（官网）
http://blog.csdn.net/anzhsoft/article/details/19607841（翻译）