Azure Messaging-ServiceBus Messaging消息队列技术系列5-重复消息：at-least-once at-most-once

xuyangus

　　上篇博客中，我们用实际的业务场景和代码示例了Azure Messaging-ServiceBus Messaging对复杂对象消息的支持和消息的持久化：
　　Azure Messaging-ServiceBus Messaging消息队列技术系列4-复杂对象消息是否需要支持序列化和消息持久化
　　本文中我们主要研究并介绍Azure Messaging对重复消息的支持。
　　MessageReceiver 对象创建时可以指定消息接收模式： ReceiveAndDelete 和 PeekLock （默认），其中：
　　1. 使用 ReceiveAndDelete 模式时，接收是单步操作，即当 Service Bus 收到请求时，它将消息标记为“正在使用”，然后将其返回给应用程序。ReceiveAndDelete 模式是最简
　　单的模型，并且最适合在出现故障时应用程序能够容许不处理消息的场景。理解此模式时，可考虑这种情况：使用者发出了接收请求，但在处理消息之前发生崩溃。由于 Service B
　　us已将消息标记为“正在使用”，因此当应用程序重新启动并重新开始使用消息时，它就会错过在崩溃前已使用的消息。
　　2. 在 PeekLock 模式下，接收变成两阶段操作，因此可以支持不能容许错过消息的应用程序。当 Service Bus 收到请求时，它会找到下一条要使用的消息，将其锁定以防止其他使
　　用者接收它，然后将其返回给应用程序。应用程序完成消息处理（或将消息可靠地存储以便将来处理）后，会对收到的消息调用 Complete 以完成接收过程的第二阶段。当Service
　　Bus 看到 Complete 时，会将该消息标记为“正在使用”。另外两个结果也是可能的。第一个结果，如果由于某种原因应用程序无法处理该消息，它可以对收到的消息Abandon（而
　　不是 Complete）。这将导致 Service Bus 解锁该消息，并使该消息可以重新被同一使用者或其他竞争的使用者接收。第二个结果，即存在与锁定关联的超时，如果应用程序在锁
　　定超时到期前无法处理改消息（例如，应用程序崩溃）则 Service Bus 将解锁该消息并使其可以重新被接收。如果应用程序在处理该消息后崩溃，但此时尚未发出 Complete 请
　　求，则在应用程序重新启动时，该消息将重新传递给应用程序。这通常称为“至少一次”处理。这意味着每条消息都将至少处理一次，但在某些情况下可能会重新传递同一消息。如果
　　方案不能容许重复处理，则需要在应用程序中添加检测重复项的逻辑。这可以基于消息的 MessageId 属性来实现。此属性的值在传递尝试过程中保持不变。这称为“恰好一次”处、
　　理。
　　接下来，我们通过Code show一下消息的重复发送和重复接收。
　　消息重复发送：同一个消息BrokeredMessage发送两次



/// <summary>
/// 发送消息
/// </summary>
private static void MessageMultiSendTest()
{
var sbUtils = new ServiceBusUtils();
//创建队列
sbUtils.CreateQueue(queueName, false);
//多次发送消息到OrderQueue
var queueSendClient = sbUtils.GetQueueClient(queueName);
var order = CreateSalesOrder(1);
var message = sbUtils.Create(order);
queueSendClient.Send(message);
queueSendClient.Send(message);
Console.WriteLine("Send Completed!");
}
　　实际执行过程中是出错的：

　　由此可以得出：
　　Azure Messaging 不支持同一个消息发送多次，必须通过new多个BrokeredMessage实例实现。
　　同时，消息的唯一性由消息的MessageID来标识！
　　PeekAndLock模式下消息的重复接收：
　　接收模式PeekAndLock，同一个队列，第一个Consumer接收消息，但是不Complete；然后第二个Consumer继续接收消息，此时第一个Consumer未Complete的消息有一个
　　TTL，在TTL时间区间之内，第二个Consumer可以继续接收当前队列未锁定的消息，当TTL时间到达后，释放第一个Consumer锁定的消息，第二个Consumer读取到了第一个
　　Consumer未Complete的消息。
　　The duration of a peek lock; that is, the amount of time that the message is locked for other receivers. The maximum value for LockDuration is 5 minutes;
　　the default value is 1 minute.



/// <summary>
/// 接收消息
/// </summary>
private static void MessageReceive()
{
int index = 0;            
var sbUtils = new ServiceBusUtils();
var queueReveiveClient1 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
for (int i = 0; i < 10; i++)
{
var msg = queueReveiveClient1.Peek();
Console.WriteLine(string.Format("Received {0} MessageID: {1}", i, msg.MessageId));
}
var queueReveiveClient2 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
for (int i = 0; i < 10; i++)
{
var msg = queueReveiveClient2.Receive();
Console.WriteLine(string.Format("Second received {0} MessageID: {1}", i, msg.MessageId));
msg.Complete();
}
////删除队列
//sbUtils.DeleteQueue(queueName);

Console.WriteLine("Receive Completed!");
}
　　第一个队列Consumer使用Peek模式接收到消息，只是取出消息，不从消息队列中移出。
　　第二个队列Consumer使用Receive模式同样可以接收消息。

　　如果两个队列Consumer都使用Receive模式接收消息，只有第一个Consumer可以接收到，第二个Consumer则接收不到，一直在等待消息的入队！



/// <summary>
/// 接收消息
/// </summary>
private static void MessageReceive()
{
int index = 0;            
var sbUtils = new ServiceBusUtils();
var queueReveiveClient1 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
for (int i = 0; i < 10; i++)
{
var msg = queueReveiveClient1.Receive();
Console.WriteLine(string.Format("Received {0} MessageID: {1}", i, msg.MessageId));
}
var queueReveiveClient2 = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.PeekLock);
for (int i = 0; i < 10; i++)
{
var msg = queueReveiveClient2.Receive();
Console.WriteLine(string.Format("Second received {0} MessageID: {1}", i, msg.MessageId));
msg.Complete();
}
////删除队列
//sbUtils.DeleteQueue(queueName);

Console.WriteLine("Receive Completed!");
}

　　因为消息已经被第一个Consumer消费。
　　通过本篇我们了解了Azure Messaging-ServiceBus Messaging对重复消息的处理机制。
　　周国庆
　　2017/3/16