Azure Messaging-ServiceBus Messaging消息队列技术系列4-复杂对象消息是否需要支持序列化和消息持久化

qq489498494

　　在上一篇中，我们介绍了消息的顺序收发保证：
　　Azure Messaging-ServiceBus Messaging消息队列技术系列3-消息顺序保证
　　在本文中我们主要介绍下复杂对象消息是否需要支持序列化以及消息的持久化。
　　在实际的业务应用开发中，我们经常会将复杂业务对象放到消息里面，实现异构系统之间的集成、模块间的解耦等等。
　　同时，我们还比较关注消息队列服务是否支持消息的持久化，消息队列如果宕机后持久化的消息是否可以还原？
　　在Azure Messaging的官方说明中，没有特地的介绍复杂对象消息是否需要支持序列化的要求，但是，我们在上篇博文中，有个消息创建方法，as following,
　　BrokeredMessage类的构造函数：



//
// Summary:
//     Constructor that creates a BrokeredMessage from a given object using the
//     provided XmlObjectSerializer
//
// Parameters:
//   serializableObject:
//     The serializable object.
//
//   serializer:
//     The serializer object.
//
// Exceptions:
//   System.ArgumentNullException:
//     Thrown when null serializer is passed to the method with a non-null serializableObject
//
// Remarks:
//     You should be aware of the exceptions that their provided Serializer can
//     throw and take appropriate actions. Please refer to for a possible list of
//     exceptions and their cause.
public BrokeredMessage(object serializableObject, XmlObjectSerializer serializer);
　　看来消息的构造，支持动态传入XmlObjectSerializer, so,



1         /// <summary>
2         /// 构造消息
3         /// </summary>
4         /// <param name="serializableObject">可序列化的对象</param>
5         /// <returns>消息</returns>
6         public BrokeredMessage Create(Object serializableObject)
7         {
8             var serializer = new DataContractSerializer(serializableObject.GetType(),
9                 new DataContractSerializerSettings() { IgnoreExtensionDataObject = true, PreserveObjectReferences = true });
10             var message = new BrokeredMessage(serializableObject, serializer);
11             message.Properties.Add("Type", serializableObject.GetType().ToString());
12
13             return message;
14         }
　　接下来，我们用上一篇中的代码，做一个复杂对象消息收发的测试，我们还是用上次的SalesOrder类，但是增加一个SalesOrderItem集合和双向关联，来描述销售订单和销售订单明细的的1：n的业务领域模型。



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; }
private List<SalesOrderItem> items;
/// <summary>
/// 销售订单明细
/// </summary>
public List<SalesOrderItem> Items
{
get
{
if (items == null)
items = new List<SalesOrderItem>();
return items;
}
set
{
items = value;
}
}
}
}


using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace AzureMessaging.FIFO
{
/// <summary>
/// 销售订单明细
/// </summary>
public class SalesOrderItem
{
/// <summary>
/// 标识
/// </summary>
public string ID { get; set; }
/// <summary>
/// 客户ID
/// </summary>
public int CustomerID { get; set; }
/// <summary>
/// 所属的销售订单ID
/// </summary>
public string SalesOrderID
{
get
{
if (Order != null)
return Order.OrderID;
return string.Empty;
}
}
/// <summary>
/// 所属的销售订单
/// </summary>
public SalesOrder Order { get; set; }
}
}
　　创建销售订单实例类方法：



  private static SalesOrder CreateSalesOrder(int i)
{
var order = new SalesOrder() { OrderID = i.ToString(), Code = "SalesOrder_" + i, CreateTime = DateTime.Now, ProductID = 17967, TotalPrice = new decimal(19999) };
order.Items.Add(new SalesOrderItem() { ID = Guid.NewGuid().ToString(), Order = order, CustomerID = 1234567 });
return order;
}
　　在构造SalesOrder和SalesOrderItems时，我们做了双向关联。
消息顺序收发测试：



1 using Microsoft.ServiceBus.Messaging;
2 using System;
3 using System.Collections.Generic;
4 using System.Linq;
5 using System.Text;
6 using System.Threading.Tasks;
7
8 namespace AzureMessaging.FIFO
9 {
10     class Program
11     {
12         private static readonly string queueName = "OrderQueue";
13         static void Main(string[] args)
14         {
15             MessageSend();
16             Console.ReadKey();
17
18             MessageReceive();
19             Console.ReadKey();
20         }
21
22         /// <summary>
23         /// 发送消息
24         /// </summary>
25         private static void MessageSend()
26         {
27             var sbUtils = new ServiceBusUtils();
28
29             //创建队列
30             sbUtils.CreateQueue(queueName, false);
31
32             //顺序发送消息到OrderQueue
33             var queueSendClient = sbUtils.GetQueueClient(queueName);
34             for (int i = 0; i < 10; i++)
35             {
36                 var order = CreateSalesOrder(i);
37                 var message = sbUtils.Create(order);
38                 queueSendClient.Send(message);
39                 Console.WriteLine(string.Format("Send {0} MessageID: {1}", i, message.MessageId));
40             }
41
42             Console.WriteLine("Send Completed!");
43         }
44
45         /// <summary>
46         /// 接收消息
47         /// </summary>
48         private static void MessageReceive()
49         {
50             int index = 0;
51             BrokeredMessage msg = null;
52             var sbUtils = new ServiceBusUtils();
53             var queueReveiveClient = sbUtils.GetReceiveQueueClient(queueName, ReceiveMode.ReceiveAndDelete);
54             while ((msg = queueReveiveClient.Receive(TimeSpan.FromMilliseconds(3))) != null)
55             {
56                 Console.WriteLine(string.Format("Received {0} MessageID: {1}", index, msg.MessageId));
57                 index++;
58             }
59
60             ////删除队列
61             //sbUtils.DeleteQueue(queueName);
62
63             Console.WriteLine("Receive Completed!");
64         }
65
66         private static SalesOrder CreateSalesOrder(int i)
67         {
68             var order = new SalesOrder() { OrderID = i.ToString(), Code = "SalesOrder_" + i, CreateTime = DateTime.Now, ProductID = 17967, TotalPrice = new decimal(19999) };
69             order.Items.Add(new SalesOrderItem() { ID = Guid.NewGuid().ToString(), Order = order, CustomerID = 1234567 });
70
71             return order;
72         }
73     }
74 }

　　可以看出，复杂对象消息只要指定适当的XmlObjectSerializer，即可。
　　在双向引用这种领域模型的设计场景下，我们配置了PreserveObjectReferences = true



var serializer = new DataContractSerializer(serializableObject.GetType(),
new DataContractSerializerSettings() { IgnoreExtensionDataObject = true, PreserveObjectReferences = true });
　　解决了序列化时循环引用的问题。
关于消息的持久化，Azure messaging有官方的说明：所有的队列都是持久化的，持久化存储是SQL Server，不提供内存中的消息队列。
　　毕竟是PaaS层的消息队列服务，消息的持久化和高可用性微软还是有保障的。

　　本篇中我们介绍并验证了Azure Messaging Service Bus复杂对象消息是否需要支持序列化和消息持久化，下一篇我们继续介绍消息的重复发送问题。
　　周国庆
　　2017/3