上篇博文中我们介绍了Azure Messaging的重复消息机制、At most once 和At least once.
Azure Messaging-ServiceBus Messaging消息队列技术系列5-重复消息：at-least-once at-most-once
本文中我们主要研究并介绍Azure Messaging的消息回执机制：实际应用场景：
同步收发场景下，消息生产者和消费者双向应答模式，例如：张三写封信送到邮局中转站，然后李四从中转站获得信，然后在写一份回执信，放到中转站，然后张三去取，当然张三写信的时候就得写明回信地址。还
有，生成订单编号场景，发送一个生成订单编号的消息，消息消费者接收生成订单编号的消息，并通过消息回执返回。
Azure Messaging的消息回执机制主要通过：基于带会话的Queue/Topic、SessionId、ReplyTo属性来实现
在代码实现中，我们需要：
1. 两个工作线程，一个线程用于消息发送和接收回执消息，一个线程用于消息接收和发送消息回执。
2. 一个会话标识：ReceiptSession
3. 两个队列Queue：RequestQueue：发送消息、接收消息，ResponseQueue：发送回执消息，接收回执消息。
直接Show Code：
首先，我们在ServiceBusMQManager增加一个线程安全的创建带回话的QueueClient方法：
private static object syncObj = new object(); /// <summary>
/// 获取要求会话带Session的QueueClient /// </summary>
/// <param name="queueName">队列名称</param>
/// <returns>QueueClient</returns>
public QueueClient GetSessionQueueClient(string queueName)
{ var namespaceClient = NamespaceManager.Create(); if (!namespaceClient.QueueExists(queueName))
{ lock (syncObj)
{ if (!namespaceClient.QueueExists(queueName))
{ var queue = new QueueDescription(queueName) { RequiresSession = true };
namespaceClient.CreateQueue(queue);
}
}
} return QueueClient.Create(queueName, ReceiveMode.ReceiveAndDelete);
}
然后我们定义一些常量：
private static readonly string ReplyToSessionId = "ReceiptSession"; const double ResponseMessageTimeout = 20.0; private static readonly string requestQueueName = "RequestQueue"; private static readonly string responseQueueName = "ResponseQueue";
实现发送并接收回执消息的方法：
/// <summary>
/// 发送并接收回执消息 /// </summary>
/// <param name="bills"></param>
public static void SendMessage()
{ var manager = new ServiceBusUtils(); var responseClient = manager.GetSessionQueueClient(responseQueueName); var requestClient = manager.GetSessionQueueClient(requestQueueName); var messsageReceiver = responseClient.AcceptMessageSession(ReplyToSessionId); var order = CreateSalesOrder(1); //发送消息
var message = new BrokeredMessage(order);
message.Properties.Add("Type", order.GetType().ToString());
message.SessionId = ReplyToSessionId;
message.MessageId = "OrderMessage001";
message.ReplyTo = responseQueueName;
requestClient.Send(message);
Console.WriteLine("Send message: " + message.MessageId + ", SalesOrder ID: " + order.OrderID); //接收消息回执
var receivedMessage = messsageReceiver.Receive(TimeSpan.FromSeconds(ResponseMessageTimeout * 2)); var receivedOrder = receivedMessage.GetBody<SalesOrder>();
Console.WriteLine("Receive receipt message: " + receivedMessage.MessageId + ", SalesOrder ID: " + receivedOrder.OrderID);
messsageReceiver.Close();
}
实现接收消息并发送回执方法：
1 /// <summary> 2 /// 接收消息并回执 3 /// </summary> 4 public static void ReceiveMessage() 5 { 6 var manager = new ServiceBusUtils(); 7 8 var requestClient = manager.GetSessionQueueClient(requestQueueName); 9 var session = requestClient.AcceptMessageSession();10 var requestMessage = session.Receive();11 12 if (requestMessage != null)13 {14 var receivedOrder = requestMessage.GetBody<SalesOrder>();15 Console.WriteLine("Receive message: " + requestMessage.MessageId + ", SalesOrder ID: " + receivedOrder.OrderID);16 17 var responseMessage = new BrokeredMessage(receivedOrder);18 responseMessage.Properties.Add("Type", receivedOrder.GetType().ToString());19 responseMessage.ReplyToSessionId = ReplyToSessionId;20 responseMessage.MessageId = "ResponseOrderMessage001";21 responseMessage.SessionId = requestMessage.SessionId;22 23 //发送回执消息24 var responseClient = manager.GetSessionQueueClient(requestMessage.ReplyTo);25 responseClient.Send(responseMessage);26 Console.WriteLine("Send receipt message: " + responseMessage.MessageId + ", SalesOrder ID: " + receivedOrder.OrderID);
27 }28 }
Main方法中，启动两个工作线程：一个线程用于消息发送和接收回执消息，一个线程用于消息接收和发送消息回执。
因为涉及到Azure Messaging中队列的第一次创建，Azure Messaging是不支持多个请求同时创建同一个队列的，因此，我们两个线程间做一个简单的Task.Delay(3000).Wait();
1 static void Main(string[] args) 2 { 3 var sendTask = Task.Factory.StartNew(() => { SendMessage(); }); 4 Task.Delay(3000).Wait(); 5 var receiveTask = Task.Factory.StartNew(() => { ReceiveMessage(); }); 6 7 Task.WaitAll(sendTask, receiveTask); 8 9 Console.ReadKey();
10 }
我们看看程序输出：
Azure 服务总线中的队列：
可以看出：Azure Messaging-ServiceBus Messaging 基于带会话的Queue/Topic、SessionId、ReplyTo属性来实现消息回执机制。
http://www.cnblogs.com/tianqing/p/6607682.html
延伸阅读
- 构建混合应用方式之Azure WCF中继 2017-07-26 16:20:39.78
- 构建混合应用方式之Azure混合连接 2017-07-11 09:30:44.493
- Azure IoT 技术研究系列2-设备注册到Azure IoT Hub 2017-07-05 15:50:27.387
- Azure IoT 技术研究系列1-入门篇 2017-07-05 10:30:16.337
- Python 操作 Azure Blob Storage 2017-06-23 15:22:14.817
- Azure Messaging-ServiceBus Messaging消息队列技术系列5-重复消息：at-least-once at-most-once 2017-06-20 16:34:43.907
- Azure经典门户创建VM，如何设置使用静态IP地址？ 2017-06-14 16:55:43.723
- 微软.NET年芳15：我在Azure上搭建Photon服务器(C#.NET) 2017-06-08 14:39:18.22
- Azure Messaging-ServiceBus Messaging消息队列技术系列7-消息事务 2017-06-06 14:53:17.143
- 在Azure China用自定义镜像创建Azure VM Scale Set 2017-06-02 16:12:57.997
构建混合应用方式之Azure混合连接