Windows Azure: Service Bus Brokered Messaging DeadLetterQueue 使用详解

[code] string queueName = "MyQueue";

NamespaceManager namespaceClient = NamespaceManager.Create();

if (namespaceClient.QueueExists(queueName))

{

namespaceClient.DeleteQueue(queueName);

}

QueueDescription queueDescription = new QueueDescription(queueName);

namespaceClient.CreateQueue(queueDescription);

MessagingFactory factory = MessagingFactory.Create();

QueueClient queueClient = factory.CreateQueueClient(queueName);

CreateAndSendOrderMessage(Guid.NewGuid().ToString(), "Beijing", queueClient);

CreateAndSendOrderMessage(Guid.NewGuid().ToString(), "Dalian", queueClient);

CreateAndSendOrderMessage(Guid.NewGuid().ToString(), "Guangzhou", queueClient);

Console.WriteLine();

Console.WriteLine("Press [Enter] to delete queue and exit.");

Console.ReadLine();

namespaceClient.DeleteQueue(queueName);

factory.Close();

queueClient.Close();

[/code]
[/code]

[code]
[code] private static void CreateAndSendOrderMessage(string orderId, string orderRegion, QueueClient sender)

{

 var message = new BrokeredMessage();

 message.Properties.Add("OrderId", orderId);

 message.Properties.Add("OrderRegion", orderRegion);

 Console.WriteLine("Sending message of order region{0}.", message.Properties["OrderRegion"]);

 sender.Send(message);

}

[/code]
[/code]

接收消息：

[code][code] string queueName = "MyQueue";

MessagingFactory factory = MessagingFactory.Create();

QueueClient queueClient = factory.CreateQueueClient(queueName, ReceiveMode.PeekLock);

Console.WriteLine("Reading messages from queue...");

BrokeredMessage message = null;

while ((message = queueClient.Receive(TimeSpan.FromSeconds(2))) != null)

{

 string orderRegion = message.Properties["OrderRegion"].ToString();

 if (validatedRegion.Contains(orderRegion))

{

 Console.WriteLine("Process the message whose order region is{0}", orderRegion);

 ProcessMessage(message);

 message.Complete();

}

 else

{

 Console.WriteLine("DeadLetter the message whose order region is{0}", orderRegion);

 message.DeadLetter("Invalid Region", string.Format("The order region{0} is invalid", orderRegion));

}

 message.Dispose();

}

queueClient.Close();

[/code]
[/code]
处理DeadLetter队列中的消息：

[code]
[code] string deadLetterQueuePath = QueueClient.FormatDeadLetterPath(queueName);

QueueClient deadletterQueueClient = factory.CreateQueueClient(deadLetterQueuePath, ReceiveMode.PeekLock);

Console.WriteLine("Reading messages from deadletter queue...");

BrokeredMessage messageInDeadletterQueue = null;

while ((messageInDeadletterQueue = deadletterQueueClient.Receive(TimeSpan.FromSeconds(2))) != null)

{

 LogDeadletterMessage(messageInDeadletterQueue);

 messageInDeadletterQueue.Complete();

 messageInDeadletterQueue.Dispose();

}

deadletterQueueClient.Close();

[/code]
[/code]
过期消息

假设我们在创建队列或者Subscription时，将默认的消息存活时间设置为5s（当然也可以给每个消息单独设置TimeToLive），并且将EnableDeadLetteringOnMessageExpiration设置为true，如果为false，则过期消息将被直接删除。

[code][code] string queueName = "MyQueue";

NamespaceManager namespaceClient = NamespaceManager.Create();

if (namespaceClient.QueueExists(queueName))

{

namespaceClient.DeleteQueue(queueName);

}

QueueDescription queueDescription = new QueueDescription(queueName)

{

 DefaultMessageTimeToLive = TimeSpan.FromSeconds(5),

 EnableDeadLetteringOnMessageExpiration = true

};

namespaceClient.CreateQueue(queueDescription);

[/code]
[/code]
那在消息发送以后，等5s我们再去正常接收消息，会发现没有任何消息，消息因超期被移入了DeadLetter队列中，接收消息代码同显示调用DeadLetter。测试结果如图所示：

消息发送：



消息接收：



接收次数超过MaxDeliveryCount值

我们将Queue或Subscription的MaxDeliveryCount属性设置为2，那么一旦消息被接收的次数到达2次以后将会被移入DeadLetter队列。

[code][code] string queueName = "MyQueue";

NamespaceManager namespaceClient = NamespaceManager.Create();

if (namespaceClient.QueueExists(queueName))

{

namespaceClient.DeleteQueue(queueName);

}

QueueDescription queueDescription = new QueueDescription(queueName)

{

 MaxDeliveryCount = 2

};

namespaceClient.CreateQueue(queueDescription);

[/code]
[/code]
接收消息：

[code][code] string queueName = "MyQueue";

MessagingFactory factory = MessagingFactory.Create();

QueueClient queueClient = factory.CreateQueueClient(queueName, ReceiveMode.PeekLock);

Console.WriteLine("Reading messages from queue...");

BrokeredMessage message = null;

while ((message = queueClient.Receive(TimeSpan.FromSeconds(2))) != null)

{

 Console.WriteLine("Order Id:{0}", message.Properties["OrderId"]);

 message.Abandon();

 message.Dispose();

}

queueClient.Close();

[/code]
[/code]
从DeadLette队列中收取消息代码同上。

测试结果：

从图中我们可以看到，总共3个订单消息，每个消息被接收了两次，然后从DeadLetter队列中就能收取到这3条消息。



Filter Evaluation Exception

快崩溃了，一直没能模拟出Filter Evaluation Exception，所以无法为大家提供例证。后期我再补全并上传代码。哪位好心朋友如果清楚Filter Evaluation Exception如何模拟请给我留言，谢谢。