Azure Messaging-ServiceBus Messaging消息队列技术系列3-消息顺序保证

上一篇:Window Azure ServiceBus Messaging消息队列技术系列2-编程SDK入门  http://www.cnblogs.com/tianqing/p/5944573.html

介绍了Azure Service Bus的编程SDK（主要的编程接口）

本文中我们以实际的使用场景来说明Azure Messaging是否支持以及如何编码实现：消息的收发顺序保证。

消息的收发在实际业务中往往是有顺序的：发送时1-2-3-4-5，接收时也必须是1-2-3-4-5，即FIFO特性。

在本文的Demo中，我们模拟销售订单消息队列异步处理场景，消息体是一条SalesOrder，顺序发送，顺序接收。

1. 我们还是使用上篇博客中在Windows Azure的Portal上建立好的NameSpace：servicebustest

销售订单队列名称：OrderQueue

2.简单封装一个Service Bus的工具类：ServiceBusUtils: 用于创建队列、删除队列、创建QueueClient、创建BrokerdMessage

using Microsoft.ServiceBus;
 Microsoft.ServiceBus.Messaging;
 System;
 System.Collections.Generic;
 System.Linq;
 System.Runtime.Serialization;
 System.Text;
 System.Threading.Tasks;

namespace AzureMessaging.FIFO
{
    /// <summary>
    /// ServiceBus工具类
    </summary>
    class ServiceBusUtils
    {
        //Namespace名称
        private static readonly string namespaceName = "servicebustest";

        <summary>
         创建队列
        </summary>
        <param name="queueName">队列名称</param>
        <param name="isSession">是否支持会话</param>
        public void CreateQueue(string queueName,bool isSession = true)
        {
            var namespaceClient = NamespaceManager.Create();
            if (namespaceClient.QueueExists(queueName))
            {
                namespaceClient.DeleteQueue(queueName);
            }

            var queue = new QueueDescription(queueName) { RequiresSession = isSession };
            namespaceClient.CreateQueue(queue);
        }

         删除队列
        </param>        
        void DeleteQueue(string queueName)
        {
             (namespaceClient.QueueExists(queueName))
            {
                namespaceClient.DeleteQueue(queueName);
            }
        }


         创建队列客户端
        <returns>队列客户端</returns>
        public QueueClient GetQueueClient(false,ReceiveMode mode = ReceiveMode.ReceiveAndDelete)
        {
            return QueueClient.Create(queueName,mode);
        }
        public QueueClient GetReceiveQueueClient( ReceiveMode.PeekLock)
        {
             构造消息
        <param name="serializableObject">可序列化的对象消息public BrokeredMessage Create(Object serializableObject)
        {
            var serializer = new DataContractSerializer(serializableObject.GetType(),new DataContractSerializerSettings() { IgnoreExtensionDataObject = true,PreserveObjectReferences = false });
            var message =  BrokeredMessage(serializableObject);
            message.Properties.Add(Type,serializableObject.GetType().ToString());

             message;
        }
    }
}

2. 示例SalesOrder实体类

 销售订单类
     SalesOrder
    {
         订单ID
        </summary>
        string OrderID { get; set; }

         订单编号
        string Code {  创建时间
        public DateTime CreateTime {  总价格
        public Decimal TotalPrice {  产品ID
        int ProductID { ; }
    }
}

3. 消息顺序发送

向OrderQueue发送10条消息订单消息，输出每条消息的顺序号以及MessageID

string queueName = OrderQueue;
     发送消息
        void MessageSend()
        {
            var sbUtils =  ServiceBusUtils();

            创建队列
            sbUtils.CreateQueue(queueName,1)">);

            顺序发送消息到OrderQueue
            var queueSendClient = sbUtils.GetQueueClient(queueName);
            for (int i = 0; i < 10; i++)
            {
                var order = new SalesOrder() { OrderID = i.ToString(),Code = SalesOrder_" + i,CreateTime = DateTime.Now,ProductID = 17967,TotalPrice = new decimal(19999) };
                var message = sbUtils.Create(order);
                queueSendClient.Send(message);
                Console.WriteLine(string.Format(Send {0} Message: {1}Send Completed!);
        }

程序输出：

4. 消息顺序接收

消费OrderQueue中的消息，验证消息的接收顺序

;
  接收消息
         MessageReceive()
        {
            int index = 0;
            BrokeredMessage msg = null;
             ServiceBusUtils();
            var queueReveiveClient = sbUtils.GetReceiveQueueClient(queueName,ReceiveMode.ReceiveAndDelete);
            while ((msg = queueReveiveClient.Receive(TimeSpan.FromMilliseconds(3))) != )
            {
                Console.WriteLine(Received {0} Message: {1};
            }

            /删除队列
            sbUtils.DeleteQueue(queueName);

            Console.WriteLine(Receive Completed!);
        }

程序输出：

可以看出，Azure Messaging中ServiceBus对消息的收发是有顺序保证的。

下一篇我们继续其他特性的验证和介绍。

 

周国庆

2017/3/9