利用 Amazon Web Services 集成企业应用程序--使用 Amazon SQS 发送 XML 消息(3)

发布时间:2012-04-05 14:04 作者:佚名来源:中国信息安全博士网点击:加载中...次

　　表 3. Amazon SMS 消息队列

队列名	可见性超时	描述
POQueue	30 秒	从分销商发送给制造商的购买订单消息
OSQueue	30 秒	从制造商发送给分销商的订单汇总消息

　　分销商实现

　　分销商应用程序包含表 4 中的业务实体。

　　表 4. 分销商应用程序中的业务实体

类	描述
CompanyAddressEntity	包含一个客户或业务地址
CustomerEntity	包含一个客户
OrderEntity	包含一个客户订单
OrderDetailEntity	包含客户订单的商品细节
InventoryItemEntity	包含库存商品
PurchaseOrderEntity	包含购买订单
PurchaseOrderDetailEntity	包含购买订单的商品细节

　　对于 Microsoft .NET，BaseSQSDataProvider 类还为 Amazon SQS 消息提供一个实用程序类。BaseDbProvider 为使用 MySQL 数据库提供一个实用程序类。

　　分销商订单管理系统

　　分销商订单管理客户端不是本文的重点，但该客户端是相当简单的。它是用 C# 编写的，并且是一个 n 层应用程序，其中所有业务和数据都与表示层分离。所有表单绑定都使用 Microsoft .NET 强大的对象绑定功能完成。所有下拉列表、网格视图和表单字段都绑定到业务对象。Windows® 表单使用的代码量很少，因为业务逻辑都保存在业务逻辑层。

　　图 5 显示了这个客户端。

　　图 5. 分销商订单管理客户端

　　分销商订单完成服务

　　OrderFulfillmentService 类负责处理客户订单，并且根据在 App.config 文件中指定的轮询间隔进行处理。这个服务获取尚未处理的订单列表。对于列表中的每个订单，以下操作会发生在相应的方法中：

　　检查是否可以配送订单商品 —— 即库存是否充足(ProcessPendingOrders 方法)。

　　如果能够配送订单商品就处理该订单(CanShip 方法)。

　　如果不能配送订单商品，就推迟订单，并向制造商发送购买订单(ProcessBackorder 方法)。

　　清单 9 显示了 ProcessPendingOrders() 方法。

　　清单 9. ProcessPendingOrders() 方法

				
	public int ProcessPendingOrders()
	{
		int itemsProcessed = 0;

		// get orders not yet shipped
		CustomerOrderFactory factory = new CustomerOrderFactory();
		IList<OrderEntity> orders = factory.GetOrdersNotYetShipped();

		// iterate through all orders not processed
		IEnumerator<OrderEntity> ordersEnum = orders.GetEnumerator();
		while (ordersEnum.MoveNext()) {
			// get next order
			OrderEntity curOrder = ordersEnum.Current;
			Console.WriteLine(string.Format("Processing Order '{0}'...", 
				curOrder.Id));

			// check if merchandise is available to ship
			if (this.CanShip(curOrder)) {
				// process order
				if (this.ProcessOrder(curOrder)) {
					itemsProcessed++;
				}
			}// if can ship order
			else if (!curOrder.IsBackordered){
				// set order to backordered
				if (this.ProcessBackorder(curOrder)) {
					itemsProcessed++;
				}
			} // if can't ship order (not enough merchandise)
		} // while more orders to process

		return itemsProcessed;
	}

　　为了确定订单能否被处理，将根据订购的每种商品核查库存是否充足。清单 10 显示了 CanShip() 方法。

　　清单 10. CanShip() 方法

				
private bool CanShip(OrderEntity order)
{
bool hasMerchandise = true;            

// get items
IEnumerator<OrderDetailEntity> detailEnum = order.GetOrderItems();

// iterate through all items
while (detailEnum.MoveNext())
{
	// get current item
	OrderDetailEntity detailEntry = detailEnum.Current;

	InventoryItemEntity inventoryItem = detailEntry.Item;
	if (detailEntry.Quantity > inventoryItem.Quantity)
	{
		Console.WriteLine(
			string.Format("Order {0} - Insufficient Inventory: {1} ({2})",
			order.Id, inventoryItem.Name, inventoryItem.Id));
		hasMerchandise = false;
	} // if inventory is sufficient
} // while more entries to process

Console.WriteLine(string.Format("Order {0} - Can Ship: {1}", 
	order.Id, hasMerchandise));
return hasMerchandise;
}

　　如果 CanShip() 返回 False 并且该订单还没有推迟，那么将调用 ProcessBackorder() 方法，并且订单的 IsBackordered 属性被设置为 True。您将使用 MessageQueueFactory 创建队列项并发送购买订单消息。清单 11 显示了这个过程。

　　清单 11. ProcessBackorder() 方法

				
private bool ProcessBackorder(OrderEntity order)
{
	// set to backordered
	order.IsBackordered = true;       

	// update order
	CustomerOrderFactory factory = new CustomerOrderFactory();
	bool result = factory.UpdateOrder(order);

	if (!result) return result;

	// get purchase order xml
	string poXml = this.GetPurchaseOrderAsXml(order);            

	// create message queue
	MessageQueueFactory queueFactory = new MessageQueueFactory();
	return queueFactory.CreatePOQueueItem(poXml);
}

　　通过 GetPurchaseOrder() 方法创建 PurchaseOrderEntity 对象。对于库存不足的每种商品，将向其订单添加一个 OrderDetailEntity。清单 12 显示了这个过程。

　　清单 12. GetPurchaseOrder() 方法

				
private PurchaseOrderEntity GetPurchaseOrder(OrderEntity order)
{
PurchaseOrderEntity po = new PurchaseOrderEntity();

po.PurchaseDate = DateTime.Now;

// set company address of the Purchase Order - the Reseller
po.CompanyAddress.CompanyName = "The Widget Factory";
po.CompanyAddress.StreetAddress = "100 Main Street";
po.CompanyAddress.CityName = "Las Vegas";
po.CompanyAddress.StateName = "NV";
po.CompanyAddress.ZipCode = "89044";

// set vendor address of the Purchase Order - the Manufacturer
po.VendorAddress.CompanyName = "Widget Supplies";
po.VendorAddress.StreetAddress = "100 Main Street";
po.VendorAddress.CityName = "Orlando";
po.VendorAddress.StateName = "FL";
po.VendorAddress.ZipCode = "32801";

// while more items to process
IEnumerator<OrderDetailEntity> orderEnum = order.GetOrderItems();
while (orderEnum.MoveNext())
{
	OrderDetailEntity orderItem = orderEnum.Current;
	InventoryItemEntity inventoryItem = orderItem.Item;

	// if insufficient inventory
	if (orderItem.Quantity > inventoryItem.Quantity)
	{
		// order the number needed plus 100
		int quantityToOrder = (orderItem.Quantity - inventoryItem.Quantity) + 100;
		PurchaseOrderDetailEntity poItem = new PurchaseOrderDetailEntity();
		poItem.ItemName = inventoryItem.Name;
		poItem.ItemId = inventoryItem.Id;
		poItem.Quantity = quantityToOrder;

		// add item to po
		po.AddItem(poItem);
	}
}

return po;            
}

　　下一步就是将 PurchaseOrderEntity 系列化到 XML。Microsoft .NET Framework 提供强大的 XML 系列化功能。您可以创建 XmlSerializerNamespaces 的一个实例，以为输出文档设置 XML 名称空间。创建 XmlWriterSettings 的一个实例来控制 XML 输出。您希望从输出中忽略 XML 声明，因为它被嵌入到消息体中。您可以使用 XmlTextWriter 类将对象系列化到一个 XML 文本写器，后者将其输出写到 StringBuilder 的一个实例。最后，您可以使用 XmlSerializer 的 Serialize() 方法将 PurchaseOrderEntity 的实例系列化到 XML。清单 13 显示了这个过程。

　　清单 13. GetPurchaseOrderAsXml() 方法

				
private string GetPurchaseOrderAsXml(OrderEntity order)
{
// get purchase order
PurchaseOrderEntity po = this.GetPurchaseOrder(order); ;

StringBuilder output = new StringBuilder();

// no name space
XmlSerializerNamespaces ns = new XmlSerializerNamespaces();
ns.Add("", "");

// settings to omit 
XmlWriterSettings settings = new XmlWriterSettings();
settings.OmitXmlDeclaration = true;

XmlWriter writer = XmlTextWriter.Create(output, settings);
XmlSerializer serializer = new XmlSerializer(typeof(PurchaseOrderEntity));
serializer.Serialize(writer, po, ns);

Debug.WriteLine(output.ToString());

return output.ToString();}

　　现在，使用清单 14 中的代码将包含购买订单的消息发送到 Amazon SQS 队列。

　　清单 14. CreatePOQueueItem() 方法

				
public bool CreatePOQueueItem(String poXml)
{
MessageQueueSQSDataProvider queueFactory = DW4DataContext.MessageQueueData;
MessageQueueEntity message = new MessageQueueEntity();
message.MessageBody = poXml;
message.MessageType = MessageTypes.PurchaseOrder;
return queueFactory.InsertQueueItem(message);			
}

　　MessageQueueSQSDataProvider 的 InsertQueueItem() 方法根据消息的类型在正确的队列中创建队列项。该方法调用基础方法 SendMessage() 将消息发送到 Amazon SQS 队列，如清单 15 所示。

　　清单 15. InsertQueueItem() 方法

				
public bool InsertQueueItem(MessageQueueEntity message)
{
String queueName = null;
if (message.MessageType == MessageTypes.OrderSummary)
{
	queueName = ConfigurationSettings.AppSettings["OrderSummaryQueue"];
	return this.SendMessage(queueName, message.MessageBody);
}
else if (message.MessageType == MessageTypes.PurchaseOrder)
{
	queueName = ConfigurationSettings.AppSettings["PurchaseOrderQueue"];
	return this.SendMessage(queueName, message.MessageBody);
}

return false;            
}

　　SendMessage() 方法创建作为 Amazon API 的一部分的 SendMessageRequest。所需的参数为队列名和消息体(XML 文档)。如果 SendMessageResponse 不为 null，消息则被成功发送。接下来的几行写出了问题诊断的关键调试信息。清单 16 显示了完成后的 SendMessage() 方法。

　　清单 16. SendMessage() 方法

				
public bool SendMessage(string queueName, string messageBody) {
bool result = true;
try {
	String accessKeyId = 
		ConfigurationSettings.AppSettings["AmazonAccessKeyID"]; ;
	String secretAccessKey = 
		ConfigurationSettings.AppSettings["AmazonSecretAccessKey"];
	AmazonSQS service = new AmazonSQSClient(accessKeyId, secretAccessKey);

	// build request
	SendMessageRequest request = new SendMessageRequest();
	request.QueueName = queueName;
	request.MessageBody = messageBody;
	// send message
	SendMessageResponse response = service.SendMessage(request);
	if (response.IsSetSendMessageResult()) {
		Debug.WriteLine("Send Message Result:");
		SendMessageResult sendMessageResult = response.SendMessageResult;
		if (sendMessageResult.IsSetMessageId()) {
			Debug.WriteLine(String.Format("\tMessageId: {0}", 
				sendMessageResult.MessageId));
		}
		if (sendMessageResult.IsSetMD5OfMessageBody()) {
			Debug.WriteLine("\tMD5 Of Message Body: ", 
				sendMessageResult.MD5OfMessageBody);
		}
	}
	if (response.IsSetResponseMetadata()) {
		Debug.WriteLine("Response Metadata:");
		ResponseMetadata responseMetadata = response.ResponseMetadata;
		if (responseMetadata.IsSetRequestId()) {
			Debug.WriteLine(String.Format("\tRequest Id: {0}", 
				responseMetadata.RequestId));
		}
	}
}
catch (AmazonSQSException ex) {
	Debug.WriteLine("Caught Exception: " + ex.Message);
	Debug.WriteLine("Response Status Code: " + ex.StatusCode);
	Debug.WriteLine("Error Code: " + ex.ErrorCode);
	Debug.WriteLine("Error Type: " + ex.ErrorType);
	Debug.WriteLine("Request ID: " + ex.RequestId);
	Debug.WriteLine("XML: " + ex.XML);
	result = false;
}
return result;
}

　　分销商库存管理服务

　　InventoryManagementService 负责管理库存并处理来自制造商的订单汇总。类似于 OrderFulfillmentService，该服务使用在 App.config 文件中指定的轮询时间间隔。将发生以下行为：

　　ProcessIncomingMerchandise() 方法首先获取接收到但尚未处理的所有订单汇总的列表(OrderSummaryEntity)。

　　InventoryManagementService 方法获取来自 Amazon SQS OSQueue 的消息。

　　然后，对每个未处理的商品发货调用 ProcessOrderReceipt() 方法。

　　清单 17 显示了 ProcessIncomingMerchandise() 方法如何获取列表。

　　清单 17. ProcessIncomingMerchandise() 方法

				
public int ProcessIncomingMerchandise()
{
int itemsProcessed = 0;

OrderSummaryFactory osFactory = new OrderSummaryFactory();
IList<OrderSummaryEntity> orders = osFactory.GetOrderSummariesToProcess();

// iterate through all order summaries
IEnumerator<OrderSummaryEntity> orderEnum = orders.GetEnumerator();
while (orderEnum.MoveNext())
{
	// get current order summary
	OrderSummaryEntity order = orderEnum.Current;

	// process order summary received
	if (this.ProcessOrderReceipt(order))
	{
		itemsProcessed++;
	}
}

Debug.WriteLine(String.Format("Orders Processed: {0}", itemsProcessed));
return itemsProcessed;
}

　　ProcessOrderReceipt()方法遍历订单汇总中的每个项(OrderSummaryDetailEntity),并增加接收到的商品的库存数量。清单 18 显示了这个方法。

　　清单 18. ProcessOrderReceipt() 方法

				
private bool ProcessOrderReceipt(OrderSummaryEntity order)
{
	if (order == null) return false;

	bool result = true;

	// add to inventory
	InventoryFactory inventoryFactory = new InventoryFactory();
	
	// iterate through all items in order summary
	IEnumerator<OrderSummaryDetailEntity> itemsEnum = order.Items.GetEnumerator();
	while (itemsEnum.MoveNext())
	{
		// get current item
		OrderSummaryDetailEntity orderItem = itemsEnum.Current;
		
		// get item
		int itemId = orderItem.ItemId;
		InventoryItemEntity item = inventoryFactory.GetInventoryItem(itemId);

		// increase inventory
		item.Quantity = item.Quantity + orderItem.Quantity;
		result = inventoryFactory.UpdateInventoryItem(item);
		if (!result) break;
	}

	if (result)
	{
		MessageQueueFactory queueFactory = new MessageQueueFactory();
		queueFactory.DeleteQueueItem(order.QueueItem);
	}

	return result;
}

(责任编辑：)