异步延迟谷歌应用引擎中的多个任务

Question

我有一个带有 3 个参数的 tasklet，一个带有 id 和标题的字典，一个用户键，以及数据库中已有的记录列表。 函数 defer_fetch 在 for 循环中被调用 N 次。

@ndb.tasklet
def defer_fetch(data, user_key, already_inserted):
    if data['_id'] not in already_inserted:
         document_key = yield Document.fetch_or_create(data)
         yield defer_fetch_document(user_key, document_key)
    else:
         document_key = alread_inserted[data['_id']]
    raise ndb.Return(document_key)

@ndb.tasklet
def defer_fetch_document(user_key, document_key):
    deferred.defer(process_document, user_key, document_key, _queue="process-documents")
    raise ndb.Return(True)

document.fetch_or_create 的代码在所有 defer_fetch 调用中并行执行，但调用 fetch_document 不是，如附件所示

如何让 defer_fetch_document 也并行运行？

Answer 1

据我所知，deferred 不能很好地与 ndb.tasklets 配合使用（deferred.defer 没有 yield，所以代码只是同步运行）。 相反，您应该直接将任务队列与它的 add_async 功能一起使用。

@ndb.tasklet
def defer_fetch_document(user_key, document_key):
    queue = taskqueue.Queue("process-documents")
    task = taskqueue.Task(url="<url for worker>", 
                          params={"document_key": document_key.urlsafe()})
    yield queue.add_async(task)  #this returns a rpc which you can yield on
    raise ndb.Return(True)

或者，生成文档键列表（从中生成任务列表），然后使用任务列表调用 add_async。