优化任务以减少交易应用程序中的CPU

Question

我设计了一个处理客户股票投资组合的交易应用程序。

我使用两种数据存储类型：

1. Stocks -包含唯一的股票名称及其每日百分比的变化。
2. UserTransactions -包含关于用户对股票的特定购买的信息:购买的价值以及当前购买的股票引用。

db.Model python模块：

class Stocks (db.Model):
stockname = db.StringProperty(multiline=True) 
dailyPercentChange=db.FloatProperty(default=1.0) 

class UserTransactions (db.Model): 
buyer = db.UserProperty() 
value=db.FloatProperty() 
stockref = db.ReferenceProperty(Stocks) 

我需要每小时更新一次数据库:更新Stocks中的每日百分比变化，然后更新UserTransactions中引用该股票的所有实体的值。

下面的python模块遍历所有股票，更新dailyPercentChange属性，并调用一个任务来遍历引用股票并更新其值的所有UserTransactions实体：

Stocks.py

# Iterate over all stocks in datastore
for stock in Stocks.all():
   # update daily percent change in datastore
   db.run_in_transaction(updateStockTxn, stock.key()) 
   # create a task to update all user transactions entities referring to this stock
   taskqueue.add(url='/task', params={'stock_key': str(stock.key(), 'value' : self.request.get ('some_val_for_stock') }) 

def updateStockTxn(stock_key):
   #fetch the stock again - necessary to avoid concurrency updates
   stock = db.get(stock_key)
   stock.dailyPercentChange= data.get('some_val_for_stock') # I get this value from outside
   ... some more calculations here ...
   stock.put()

Task.py (/task)

# Amount of transaction per task
amountPerCall=10
stock=db.get(self.request.get("stock_key")) 
# Get all user transactions which point to current stock
user_transaction_query=stock.usertransactions_set
cursor=self.request.get("cursor") 
if cursor: 
    user_transaction_query.with_cursor(cursor) 

# Spawn another task if more than 10 transactions are in datastore
transactions = user_transaction_query.fetch(amountPerCall) 
if len(transactions)==amountPerCall: 
    taskqueue.add(url='/task', params={'stock_key': str(stock.key(), 'value' : self.request.get ('some_val_for_stock'), 'cursor': user_transaction_query.cursor()  })

# Iterate over all transaction pointing to stock and update their value
for transaction in transactions: 
   db.run_in_transaction(updateUserTransactionTxn, transaction.key()) 

def updateUserTransactionTxn(transaction_key): 
   #fetch the transaction again - necessary to avoid concurrency updates
   transaction = db.get(transaction_key)
   transaction.value= transaction.value* self.request.get ('some_val_for_stock')
   db.put(transaction) 

问题：

目前，该系统运行良好，但问题是它没有很好地扩展…。我有大约100个股票与300个用户交易，我运行更新每小时。在仪表板上，我看到task.py占用了大约65%的CPU (Stock.py大约占20%-30%)，我几乎使用了应用程序引擎给我的所有6.5个免费的CPU时间。我在启用计费和支付额外CPU方面没有问题，但问题是系统…的扩展。100只股票使用6.5cpu小时是非常糟糕的。

考虑到上面提到的系统的要求，我想知道是否有比这里介绍的更好和更有效的实现(或者仅仅是一个可以帮助当前实现的小小改变)。

谢谢！！

乔尔

Answer 1

有几个明显的改进要做：

1. --您应该在第一个片段中使用keys_only查询:由于您实际上没有在任何一点上引用股票对象的属性，所以没有检索它的意义。您还可以只检索键。
2. 您可以使用Queue对象的.add方法(有文档的here )批量添加任务。这比每10个事务添加一个任务individually.
3. Your任务链更有效，但是任务最多可以运行10分钟，而10个数据存储事务可能只需要一两秒钟。相反，在请求开始时设置一个计时器，每次在循环中检查它，在接近10分钟的时间内中止并链接下一个任务。如果您希望在大量实体上迭代，则使用
4. 和游标，而不是迭代；在20个实体的小批中迭代获取。在单个实体更新中的
5. ，再次执行常规查询，但只使用键。用keys_only查询代替。
6. 是唯一在最初编写UserTransaction实体之后更新它们的任务？如果是这样的话，您可以跳过事务并批量更新它们.

最后，我建议进行总体重构:与其为每个股票启动一个新任务，不如在任务内部运行外部循环，并使用上面提到的计时器。当您链接下一个任务时，使用游标传递当前状态并拾取所中断的位置。

需要考虑的另一件事是，如果有某种方法可以重构数据，以避免需要如此多的更新。例如，您能否让UserTransaction实体引用股票实体中的一些值，以便您可以在运行时计算它们的实际值，并且只需要用更改更新单个Stock？