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

Posted 2020-09-03 I love .net

上一篇: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;
using Microsoft.ServiceBus.Messaging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.Text;
using 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);
        }

        /// <summary>
        /// 删除队列
        /// </summary>
        /// <param name="queueName">队列名称</param>        
        public void DeleteQueue(string queueName)
        {
            var namespaceClient = NamespaceManager.Create();
            if (namespaceClient.QueueExists(queueName))
            {
                namespaceClient.DeleteQueue(queueName);
            }
        }


        /// <summary>
        /// 创建队列客户端
        /// </summary>
        /// <returns>队列客户端</returns>
        public QueueClient GetQueueClient(string queueName, bool isSession = false, ReceiveMode mode = ReceiveMode.ReceiveAndDelete)
        {
            return QueueClient.Create(queueName, mode);
        }
        /// <summary>
        /// 创建队列客户端
        /// </summary>
        /// <returns>队列客户端</returns>
        public QueueClient GetReceiveQueueClient(string queueName, ReceiveMode mode = ReceiveMode.PeekLock)
        {
            var namespaceClient = NamespaceManager.Create();
            return QueueClient.Create(queueName, mode);
        }

        /// <summary>
        /// 构造消息
        /// </summary>
        /// <param name="serializableObject">可序列化的对象</param>
        /// <returns>消息</returns>
        public BrokeredMessage Create(Object serializableObject)
        {
            var serializer = new DataContractSerializer(serializableObject.GetType(), 
                new DataContractSerializerSettings() { IgnoreExtensionDataObject = true, PreserveObjectReferences = false });
            var message = new BrokeredMessage(serializableObject);
            message.Properties.Add("Type", serializableObject.GetType().ToString());

            return message;
        }
    }
}

2. 示例SalesOrder实体类

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace AzureMessaging.FIFO
{
    /// <summary>
    /// 销售订单类
    /// </summary>
    public class SalesOrder
    {
        /// <summary>
        /// 订单ID
        /// </summary>
        public string OrderID { get; set; }

        /// <summary>
        /// 订单编号
        /// </summary>
        public string Code { get; set; }

        /// <summary>
        /// 创建时间
        /// </summary>
        public DateTime CreateTime { get; set; }

        /// <summary>
        /// 总价格
        /// </summary>
        public Decimal TotalPrice { get; set; }

        /// <summary>
        /// 产品ID
        /// </summary>
        public int ProductID { get; set; }
    }
}

3. 消息顺序发送

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

private static readonly string queueName = "OrderQueue";
    /// <summary>
        /// 发送消息
        /// </summary>
        private static void MessageSend()
        {
            var sbUtils = new ServiceBusUtils();

            //创建队列
            sbUtils.CreateQueue(queueName, false);

            //顺序发送消息到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}", i, message.MessageId));
            }

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

程序输出：

4. 消息顺序接收

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

private static readonly string queueName = "OrderQueue";
 /// <summary>
        /// 接收消息
        /// </summary>
        private static void MessageReceive()
        {
            int index = 0;
            BrokeredMessage msg = null;
            var sbUtils = new ServiceBusUtils();
            var queueReveiveClient = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.ReceiveAndDelete);
            while ((msg = queueReveiveClient.Receive(TimeSpan.FromMilliseconds(3))) != null)
            {
                Console.WriteLine(string.Format("Received {0} Message: {1}", index, msg.MessageId));
                index++;
            }

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

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

程序输出：

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

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

 

周国庆

2017/3/9