Django:如何在事务中包装批量更新/插入操作？

Question

这是我的用例:

• 我有多个并行运行的芹菜任务
• 每个任务都可以批量创建或更新许多对象.为此,我正在使用django-bulk

所以基本上我使用了一个非常方便的函数insert_or_update_many:

1. 它首先执行选择
2. 如果找到对象则更新它们
3. 否则它会创建它们

但这引入了并发问题.例如:如果在步骤1期间不存在对象,则将其添加到稍后要插入的对象列表中.但在此期间可能会发生另一个Celery任务创建该对象,并且当它尝试执行批量插入时(步骤3),我收到重复条目的错误.

我想我需要在'阻塞'块中包含3个步骤.我已经阅读了有关交易的内容,并且我试图将步骤1,2,3包含在一个with transaction.commit_on_success:块内

with transaction.commit_on_success():
    cursor.execute(sql, parameters)
    existing = set(cursor.fetchall())
    if not skip_update:
        # Find the objects that need to be updated
        update_objects = [o for (o, k) in object_keys if k in existing]
        _update_many(model, update_objects, keys=keys, using=using)
    # Find the objects that need to be inserted.
    insert_objects = [o for (o, k) in object_keys if k not in existing]
    # Filter out any duplicates in the insertion
    filtered_objects = _filter_objects(con, insert_objects, key_fields)
    _insert_many(model, filtered_objects, using=using)

但这对我不起作用.我不确定我对这些交易有充分的了解.我基本上需要一个块,我可以放置几个操作,确保没有其他进程或线程访问(写入)我的数据库资源.

Answer 1

我基本上需要一个块,我可以放置几个操作,确保没有其他进程或线程访问(写入)我的数据库资源.

一般而言,Django交易不会为您提供保证.如果您来自计算机科学的其他领域,您自然会将事务视为以这种方式阻塞,但在数据库世界中,存在不同类型的锁,处于不同的隔离级别,并且它们因每个数据库而异.因此,为了确保您的事务执行此操作,您将不得不了解事务,锁和它们的性能特征,以及数据库提供的用于控制它们的机制.

然而,有一堆进程都试图锁定表以执行竞争插入并不是一个好主意.如果冲突很少发生,您可以采用乐观锁定形式,只要失败就重试事务.或许您可以将所有这些芹菜任务都指向一个进程(如果你要获得一个表锁,那么并行化并没有性能优势).

我的建议是首先忘记批量操作,然后使用Django一次做一行update_or_create.只要您的数据库具有阻​​止重复条目的约束(听起来就像它一样),这应该没有您在上面描述的竞争条件.如果性能确实证明是不可接受的,那么请研究更复杂的选项.

采用乐观并发方法意味着不是通过获取表锁来防止冲突,比如说 - 你只是正常进行,然后如果结果是问题则重试操作.在您的情况下,它可能看起来像:

while True:
    try:
        with transaction.atomic():
            # do your bulk insert / update operation
    except IntegrityError:
        pass
    else:
        break

因此,如果遇到竞争条件,结果IntegrityError将导致transaction.atomic()块回滚已经进行的任何更改,并且while循环将强制重试事务(可能大量操作现在将看到新存在的行和将其标记为更新而不是插入).

如果碰撞很少发生,这种方法可以很好地工作,如果频繁发生则非常糟糕.