在上一篇文章中，演示了一个发送者和一个消费者的情况。这一篇介绍一下多个消费者在同一个消息队列中获取消息的情况。
在有些应用当中，消费端接收到消息任务需要长时间的处理，如果等上一个消息处理完成以后再取下一个数据进行处理的话，势必会有一些延迟。在消息队列中的数据也会不断增多，延迟将越来越大。当然对于一个消费进程来说，在某些情况下可以起多个线程来处理，而在这里将介绍另一种处理方式，多个消费进程的情况。而RabbitMQ在这方面进行了很好的处理和封装，使客户程序可以很方便的使用。
其实，在代码实现上和上一遍的例子中并没有什么不同，我们只需要运行两个Server端的程序就可以了。我们可以看到，发送的消息会平均的由两个Server中的一个来处理。不会有重复。倘若一个Server程序关闭了，那之后发送的所有消息都会在还运行着的那个Server程序上处理。这可以解决我们一些需要负载均衡的场景，而且扩展非常方便，只要在运行一个Server也就是Worker就可以了，Worker之间的状态同步都免了。
当我们处理一个较长时间任务的时候，程序在处理过程中如果出现异常，或程序挂了导致消息没有处理成功，我们通常并不希望丢失该消息任务，而希望由其他Worker来处理或者等挂了的Worker重新起来以后再处理。同样，应对该场景，RabbitMQ也提供了简便的API方便我们处理。在RabbitMQ中，为了不让消息丢失，它提供了消息应答的概念。当消费者获取到了一个消息以后，需要给RabbitMQ服务一个应答的消息，告知服务我已经收到或正确处理了该消息。那么RabbitMQ可以放心的在队列中删除该消息。在上一篇的服务端的代码中
2
2
的第二个参数是true，RabbitMQ服务一旦把消息送达目标队列及认为应答了。为了演示不发送应答消息的情况，我们需要安装一个RabbitMQ的plugin：Management UI。安装这个插件比较简单。
你可以再 http://www.rabbitmq.com/management.html 了解到详细信息。
打开浏览器， http://localhost:55672，输入默认的用户名密码 guest/guest，进入主界面。界面上信息很多，我们只看最上面的一块：Queued Messages。
还是启动一个发送程序和一个接受程序。把接收程序的定义消费者的代码修改为
2
2
然后运行两个程序，发送一个消息。由于我们在接收消息的代码中没有发送消息收到的应答包，所以在刚才监控的网页上回出现如下结果：
没有回应的消息是1个。接下来我们关闭接收程序，不修改任何代码，然后再次运行接收程序。我们发现接收程序再次收到了那个原来的消息。而浏览器上显示的状态还是没有变说明还是有一个消息没有应答，因为我们第二次运行的接收程序还是没有发应答包。关掉接收程序，修改一下代码，在处理完消息的时候我们添加如下代码：
2
2
这时，我发现程序如我们所料，还是会收到一次那个消息包，但是监控网页的界面却变了：
然后，无论在运行多少次接收程序，都不会再收到该消息包了。这也就说明了，RabbitMQ真正认为该消息被正确处理了。
需要注意的是，RabbitMQ对于没有发送应答包的消息没有时间的限制，也就是说没有Timeout。RabbitMQ只会在处理消息的接收程序与RabbitMQ服务端断开连接后才会重新分配该消息。如果连接没有断，但处理程序几天没有给回应包，它也不会重新发送。所以，如果在处理程序出现异常的时候，我们可以写代码将与RabbitMQ的连接断开来实现消息的重新发送（也许会发到其他负载均衡的机器上处理）。
修改后的接收端代码如下：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
2using System;
3using System.Collections.Generic;
4using System.IO;
5using System.Linq;
6using System.Text;
7using System.Threading.Tasks;
8using System.Xml.Serialization;
9
10namespace SendService
11{
12 /// <summary>
13 /// Round-robin dispatching 循环分发
14 /// http://www.cnblogs.com/qiyebao/p/4205626.html
15 /// 一个Send和多个Receive的例子,
16 /// 还加上了ack的例子.
17 /// 优雅分发
18 /// </summary>
19 class Program
20 {
21 static void Main(string[] args)
22 {
23 var factory = new ConnectionFactory();
24 factory.HostName = "192.168.12.111";
25 factory.Port = 5672;
26 factory.UserName = "admin";
27 factory.Password = "admin";
28 //定义要发送的数据
29
30 List<RequestMessage> messages = new List<RequestMessage>();
31 for (int i = 0; i < 200; i++)
32 {
33 RequestMessage message = new RequestMessage() { MessageId = Guid.NewGuid(), Message = "Send1this is a 请求。" + i };
34 messages.Add(message);
35 }
36
37 using (var connection = factory.CreateConnection())
38 {
39 using (var channel = connection.CreateModel())
40 {
41 bool durable = true;//队列持久化
42 channel.QueueDeclare("all.sms.message", durable, false, false, null);//hello是queue的名字
43 var properties = channel.CreateBasicProperties();
44 properties.DeliveryMode = 2;//non-persistent (1) or persistent (2) 数据持久化属性
45
46 //channel.TxSelect();
47 //序列化消息对象，RabbitMQ并不支持复杂对象的序列化，所以对于自定义的类型需要自己序列化
48 foreach (var item in messages)
49 {
50 XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
51 using (MemoryStream ms = new MemoryStream())
52 {
53
54 xs.Serialize(ms, item);
55
56 byte[] bytes = ms.ToArray();
57
58 //指定发送的路由，通过默认的exchange直接发送到指定的队列中。
59
60 channel.BasicPublish("", "all.sms.message", null, bytes);
61
62 Console.WriteLine(" [x] Sent {0}", item.Message);
63 }
64 }
65 //var body = Encoding.UTF8.GetBytes(message);
66 //channel.BasicPublish("", "all.sms.message", null, body);//hello是routing key的名字
67
68 Console.ReadLine();
69 }
70 }
71 }
72 }
73
74 public class RequestMessage
75 {
76 public Guid MessageId { set; get; }
77 public string Message { set; get; }
78 }
79}
80
81
接收方代码：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
2using RabbitMQ.Client.Events;
3using System;
4using System.Collections.Generic;
5using System.IO;
6using System.Linq;
7using System.Text;
8using System.Threading;
9using System.Threading.Tasks;
10using System.Xml.Serialization;
11
12namespace ReceiveService
13{
14 /// <summary>
15 /// 循环分发
16 /// http://www.cnblogs.com/qiyebao/p/4205626.html
17 /// </summary>
18 class Program
19 {
20 static void Main(string[] args)
21 {
22 var factory = new ConnectionFactory();
23 factory.HostName = "192.168.12.111";
24 factory.Port = 5672;
25 factory.UserName = "admin";
26 factory.Password = "admin";
27 using (var connection = factory.CreateConnection())
28 {
29 using (var channel = connection.CreateModel())
30 {
31 bool durable = true;//队列持久化
32 channel.QueueDeclare("all.sms.message", durable, false, false, null);
33 var consumer = new QueueingBasicConsumer(channel);
34 channel.BasicConsume("all.sms.message", false, consumer);//需要接受方发送ack回执,删除消息
35 Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
36 while(true)
37 {
38 //阻塞函数，获取队列中的消息
39 var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作
40 //模拟长时间运行
41 Thread.Sleep(6000);
42
43 byte[] body = ea.Body;
44 XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
45 using (MemoryStream ms = new MemoryStream(body))
46 {
47
48 RequestMessage message = (RequestMessage)xs.Deserialize(ms);
49
50 Console.WriteLine("Receive a Message, Id:" + message.MessageId + " Message:" + message.Message);
51
52 }
53 //发送应答包，消息持久化时候使用
54 channel.BasicAck(ea.DeliveryTag, false);
55 }
56 }
57 }
58 }
59 }
60
61 public class RequestMessage
62 {
63 public Guid MessageId { set; get; }
64 public string Message { set; get; }
65 }
66}
67
68
需要注意的是，该功能只有在point-point模式下才有如此效果，也就是一发一接的模式下。如果是发布和订阅这种broadcasting的模式下，这种配置项的结果会有一些不同，我们下一篇再说。
