transactions - Pyramid ZopeTransactionExtension 与 celery : how can I commit transaction immediately without keep transaction after web request?

Question

这是我的情况。

我正在构建一个 RESTful web 服务，它从客户端接收数据，然后从该数据创建一个事件，然后我想将这个新事件推送到 celery 以异步处理它。

我使用 pyramid 构建 RESTful web 服务，并使用 pyramid_celery 使 pyramid 和 celery 协同工作。

这里是我的 View 的源代码:

# views.py
# This code recive data from client, then create a new record Event from this

posted_data = schema.deserialize(request.POST.mixed())

e = Event()
e.__dict__.update(posted_data)
DBSession.add(e)
transaction.commit()

print "Commited #%d" % e.id # code mark 01
fire_event.delay(e.id) # fire_event is a celery task
logging.getLogger(__name__).info('Add event #%d to tasks' % e.id)

这是我的任务的源代码:

# tasks.py
@celery.task()
def fire_event(event_id):
    e = DBSession.query(Event).get(event_id)

    if e is None:
        return

    print "Firing event %d#%s" % (event_id, e)
    logger.info("Firing event %d#%s", event_id, e)

如果我使用 pyramid 的炼金术脚手架中的默认代码，将在 code mark 01 行引发异常。像这样的异常(exception):

DetachedInstanceError: Instance <Event at ...> is not bound to a Session; ...

来自 ZopeAlchemy document ，为了避免这个异常，我像这样配置 DBSession:

# models.py
DBSession = scoped_session(sessionmaker(
                extension=ZopeTransactionExtension(keep_session=True)
            ))

现在我的问题是 Pyramid 在我的 RESTful 请求完成后与我的 MySQL 服务器保持交易。当 RESTful 请求完成后，我转到 MySQL 服务器并运行命令:

SHOW engine innodb status;

从它的结果，我看到了这个:

--TRANSACTION 180692, ACTIVE 84 sec
MySQL thread id 94, OS thread handle 0x14dc, query id 1219 [domain] [ip] [project name] cleaning up
Trx read view will not see trx with id >= 180693, sees < 180693

这意味着 Pyramid 仍然保持连接，没关系，但 Pyramid 也开始交易，这是一个问题。当我尝试使用其他工具访问我的 MySQL 服务器时，此事务会使我陷入困境。

我的问题是:

如何让 Pyramid 在 RESTful 请求完成后立即关闭事务。如果我不能，我的情况是否有其他解决方案？

非常感谢。

Answer 1

Celery 保持一种“透明地”将您的代码作为任务运行的错觉 - 您使用 @task 装饰您的函数，然后使用 my_function.delay() 并且一切都神奇地工作。

事实上，要意识到这一点有点棘手，您的代码运行在一个完全不同的进程中，可能在另一台机器上，可能是几分钟/几小时后，并且该进程中不存在 Pyramid 请求/响应周期，因此 ZopeTransactionExtension不能用于在请求完成时自动提交工作进程中的事务 - 因为没有请求，只有一个长时间运行的工作进程。

因此，不是 Pyramid 将未完成的事务挂起 - 这是您的工作进程。当您调用 e = DBSession.query(Event).get(event_id) 并且永远不会完成时，SQLAlchemy 会启动事务。

在这里，我对类似的问题写了一个更长的答案，其中包含更多细节:https://stackoverflow.com/a/16346587/320021 - 重点是为您的工作进程使用不同的 session

另一件事是，由于对象过期和其他丑陋，最好完全避免在 Pyramid 代码中使用 transaction.commit()。在 Pyramid 中，可以在请求完成后调用一个函数——我写了一个函数，它注册了一个回调，从那里调用一个 celery 任务:

from repoze.tm import after_end
import transaction

def invoke_task_after_commit(task_fn, task_args, task_kwargs):
    """
    This should ONLY be used within the web-application process managed by repoze.tm2
    otherwise a memory leak will result. See http://docs.repoze.org/tm2/#cleanup
    for more details.
    """
    t = transaction.get()  # the current transaction

    def invoke():
        task_fn.apply_async(
            args=task_args,
            kwargs=task_kwargs,
        )

    after_end.register(invoke, t)

(我从函数中删除了很多不相关的代码，所以可能会有错别字等。作为伪代码处理)