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

Question

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

我正在使用两种数据存储类型:

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

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 实体并更新它们的值:

股票.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%)，我几乎使用了 App Engine 给我的 6.5 个空闲 CPU 小时数。我在启用计费和支付额外的 CPU 方面没有问题，但问题是系统的扩展……100 只股票使用 6.5 个 CPU 小时是非常糟糕的。

我想知道，考虑到上述系统的要求，是否有比这里介绍的更好、更有效的实现(或者只是可以帮助当前实现的小改动)。

谢谢!!

乔尔

Answer 1

有几个明显的改进:

1. 您应该在第一个片段中使用 keys_only 查询:因为您在任何时候实际上都没有引用股票对象的属性，所以检索它没有意义。您也可以只检索 key 。
2. 您可以使用 Queue 对象的 .add 方法批量添加任务，已记录 here .这比单独添加任务更有效。
3. 您的任务每 10 个事务就会链接一个新任务，但任务最多可以运行 10 分钟，而 10 个数据存储区事务可能不会超过一两秒。相反，请在请求开始时设置一个计时器，并在每次循环时检查它，在接近 10 分钟限制时中止并链接下一个任务。
4. 如果您希望迭代大量实体，请使用 .fetch 和游标，而不是迭代；迭代获取 20 个实体的小批量。
5. 在单个实体更新中，您再次执行常规查询，但仅使用键。改为执行 keys_only 查询。
6. 任务是唯一会在最初编写实体后更新 UserTransaction 实体的东西吗？如果是这样，您可以跳过交易并批量更新它们。

最后，我建议进行整体重构:不要为每只股票启动一个新任务，而是在任务内运行外循环，并使用上述计时器。当您链接下一个任务时，使用游标传递当前状态并从您离开的地方继续。

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