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

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

表 3. Amazon SMS 消息队列队列名可见性超时描述 POQueue 30 秒从分销商发送给制造商的购买订单消息 OSQueue 30 秒从制造商发送给分销商的订单汇总消息分销商实现分销商应用程序包含表 4 中的业务实体。表

　　表 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;
}

(责任编辑：)