NDB 事务

事务是一项或一组必须具有原子性的操作,因此事务决不会只应用一部分。事务中的所有操作要么都应用,要么都不应用。事务的时长上限为 60 秒,在 30 秒后有 10 秒的空闲到期时间。

借助 NDB 异步 API,应用可以同时管理多个独立的事务。同步 API 可以使用 @ndb.transactional() 修饰器提供简化 API。被修饰的函数将在事务的上下文中执行。

@ndb.transactional
def insert_if_absent(note_key, note):
    fetch = note_key.get()
    if fetch is None:
        note.put()
        return True
    return False
note_key = ndb.Key(Note, note_title, parent=parent)
note = Note(key=note_key, content=note_text)
inserted = insert_if_absent(note_key, note)

如果某事务与另一个事务“冲突”,该事务将会失败;NDB 会自动重试几次失败的事务。在该事务重试时,函数可能会被调用多次。可尝试的重试次数有限(默认值为 3 次);如果该事务仍未成功,则 NDB 会引发 TransactionFailedError。您可以通过将 retries=N 传递给 transactional() 修饰器来更改重试次数。重试次数为 0 表示尝试一次该事务,但如果失败,则不重试;重试次数为 N N表示总共可以尝试 N+1 次该事务。示例:

@ndb.transactional(retries=1)
def insert_if_absent_2_retries(note_key, note):
    # do insert

在事务中,仅支持祖先查询。默认情况下,事务只能处理同一实体组中的实体(即其键具有相同“祖先实体”的实体)。

您可以通过传递 xg=True 来指定跨组(“XG”)事务(允许多达 25 个实体组):

@ndb.transactional(xg=True)
def insert_if_absent_xg(note_key, note):
    # do insert

跨组事务跨多个实体组运行,其行为类似于单组事务,但是,如果代码尝试更新多个实体组中的实体,则跨组事务不会失败。

如果函数引发一个异常,则事务会立即中止,而 NDB 会重新引发该异常,以使调用代码能够看到它。您可以通过引发 ndb.Rollback 异常(在这种情况下,函数调用会返回 None)以静默方式强制事务失败。但您无法强制事务重试。

如果某个函数并非一直需要在事务中运行,您可以将此类函数作为回调函数传递给 ndb.transaction(),而不是使用 @ndb.transactional 来修饰该函数

def insert_if_absent_sometimes(note_key, note):
    # do insert
inserted = ndb.transaction(lambda:
                           insert_if_absent_sometimes(note_key, note))

如需测试某个代码是否正在事务中运行,请使用 in_transaction() 函数。

您可以指定“事务”函数在由事务中的已有代码调用时应该具有的行为。@ndb.non_transactional 修饰器指定某函数不应在事务中运行;如果在事务中调用该函数,则它将在事务外部运行。@ndb.transactional 修饰器和 ndb.transaction 函数接受 propagation 关键字参数。例如,如果某函数应该启动一个新的独立事务,请按如下方式修饰该函数:

@ndb.transactional(propagation=ndb.TransactionOptions.INDEPENDENT)
def insert_if_absent_indep(note_key, note):
    # do insert

传播类型会与其他上下文选项和事务选项一起列出

如果您不了解相关情况,可能会将事务行为和 NDB 缓存行为混淆。如果您修改了事务内部的实体但尚未提交该事务,则 NDB 的上下文缓存将包含修改后的值,而底层数据存储区仍包含未经修改的值。

将事务性任务加入队列

您可以将任务作为 Datastore 事务的一部分加入队列,使任务仅在事务成功提交的情况下才会加入队列。如果事务未提交,则任务不会加入队列。如果事务已提交,则任务会加入队列。加入队列后,任务将不会立即执行,因此任务与事务之间不具有原子性。不过,任务加入队列后会立即重试,直至执行成功。这一点适用于在被修饰的函数执行期间加入队列的任何任务。

事务性任务非常有用,因为它们可让您将多个非 Datastore 操作组合为一个事务,进而以该事务的成功为决定因素(例如发送电子邮件以确认购买)。也可以将 Datastore 操作绑定到事务,例如,当且仅当事务成功时才提交在事务外部对实体组所做的更改。

在单个事务期间,应用插入任务队列事务性任务不能超过五个。事务性任务不能具有用户指定的名称。

from google.appengine.api import taskqueue
from google.appengine.ext import ndb
@ndb.transactional
def insert_if_absent_taskq(note_key, note):
    taskqueue.add(url=flask.url_for('taskq_worker'), transactional=True)
    # do insert