Celery+Django -- 与数据库相关任务的原子事务

Question

在我当前使用 Django、Docker-Compose 和 Celery（除其他外）的项目中，基本上传文件功能 insertIntoDatabase 是从任务中调用的，而在 views.py 中，任务是使用 @987654323 调用的@。

在databaseinserter.py中：

def insertIntoDatabase(datapoints, user, description): # datapoints is a list of dictionaries, user and description are just strings
    # convert data and upload to our database

在tasks.py中：

@app.task()
def db_ins_task(datapoints, user, description):
    from databaseinserter import insertIntoDatabase
    insertIntoDatabase(datapoints, user, description)

在views.py中：

with transaction.atomic():
    db_ins_task.delay(datapoints, user, description)

在将 Celery 引入项目之前，insertIntoDatabase 只是在 views.py 中直接调用，因此不会插入任何无效的数据点列表（即格式不正确），整个上传将被取消并回滚。但是，现在上传是在异步 celery 任务中，无效的上传不再正确回滚。既然上传是一项任务，我如何确保无效的上传仍然被取消和完全撤消？似乎 Django 1.9 有一些新的东西可能是我需要的：transaction.on_commit。然而，目前切换到 1.9 的主要问题是它似乎不是我们项目中的重要依赖项，Django-Hstore 兼容。 1.9 也处于 alpha 阶段，因此即使两者兼容，目前也不适合使用。有没有办法在 Django 1.8 中做到这一点？

我还研究了 django_transaction_barrier 并尝试使用它，但没有运气。在tasks.py中，我将任务更改为

@task(base=TransactionBarrierTask)
def db_ins_task(datapoints, user, description):
    from databaseinserter import insertIntoDatabase
    insertIntoDatabase(datapoints, user, description)

在views.py中我改变了任务执行：

with transaction.atomic():
    db_ins_task.apply_async_with_barrier(args=(data, user, description,))

但是，我的主要问题是，一旦收到任务，Celery 就会抛出一个关于意外关键字参数的错误：

worker_1   | Traceback (most recent call last):
worker_1   |   File "/usr/local/lib/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
worker_1   |     R = retval = fun(*args, **kwargs)
worker_1   |   File "/usr/local/lib/python2.7/site-packages/celery/app/trace.py", line 438, in __protected_call__
worker_1   |     return self.run(*args, **kwargs)
worker_1   | TypeError: db_ins_task() got an unexpected keyword argument '__transaction_barrier'

那么，最好的方法是什么？我是否应该继续尝试使用 django_transaction_barrier （如果我确实将它用于正确的事情）？如果是这样，我做错了什么/错过了什么会导致错误？如果没有，有什么更好的方法可以从我的数据库中清除无效上传？

Answer 1

Celery 是一个异步任务运行器，基本上，一旦任务交给 celery，它就火了，然后就忘了。您不能跨进程边界进行事务，因为 celery 将作为工作人员运行。

您始终可以运行其他任务来查找无效数据点并清理数据库。简而言之，您想要一个具有两阶段提交的分布式事务，这并不容易，因为它有自己的问题，并且不确定在 Python 中是否可用。

Answer 2

您是否考虑过将transaction.atomic 语句移到任务中？甚至插入函数本身？其中任何一个都应该工作。