Flask长时间例程

Question

我需要在我的Flask应用程序中完成一些耗时的工作，并且我希望能够异步进行。只需开始工作，然后从Javascript检查状态。

我正在尝试实现以下功能：

@app.route('/sync')
def sync():
    p = Process(target=routine, args=('abc',))
    p.start()

    return "Working..."

但是这会导致gunicorn工作进程变成僵尸进程。

如何解决？我应该使用类似Celery的东西吗？

- sashab

Answer 1

有许多选项。您可以开发自己的解决方案，使用Celery或Twisted（我确信还有更多现成的选项，但这些是最常见的）。

开发自己的内部解决方案并不难。您可以使用Python标准库中的模块：

• 当任务到达时，您在数据库中插入一个带有任务ID和状态的行。
• 然后启动一个进程执行工作，并在完成后更新行状态。
• 您可以拥有一个视图来检查任务是否完成，实际上只需检查相应位置的状态。

当然，您必须考虑要将计算结果存储在何处以及出现错误时会发生什么。

选择Celery也很容易。看起来像下面这样。 定义要异步执行的函数：

@celery.task
def mytask(data):

    ... do a lot of work ...

那么，不要直接调用任务，比如 mytask(data)，这会立即执行它，而是使用 delay 方法：

result = mytask.delay(mydata)

最后，你可以使用ready来检查结果是否可用：

result.ready()

请记住，要使用Celery，您必须运行一个外部的worker进程。

我从未了解过Twisted，所以我无法告诉您它是否比这个更加复杂（但也应该能够满足您想做的事情）。

无论如何，任何这些解决方案都应该与Flask配合得很好。要检查结果，完全不需要使用Javascript。只需创建检查状态并返回JSON的视图（您可以使用Flask的jsonify）。

- pacha

---

好的，关于将结果存储在数据库中，我确实已经做到了你所说的一切。但是，实现自己的解决方案的主要问题是，当我使用multiprocessing时，它会导致gunicorn工作进程失效。 - sashab

非常抱歉，我误读了您的问题。看起来您遇到的问题是，在返回语句之后，您的父进程正在终止，这有效地终止了您刚刚创建的子进程。为了避免这种情况发生，您只需要在父进程退出之前将子进程变成守护进程（即将其分离）python-daemon。 - pacha

当我在使用多进程时，出现了ValueError: I/O operation on closed file的错误。看起来我需要使用Popen。设置Process().daemon = True也没有帮助。 - sashab

是的，根据文档，p.daemon = True是不够的。如果你在nix平台上，我建议使用python-daemon或者使用这个代码片段，它直接调用操作系统的fork而不是multiprocessing或Popen。如果你在Windows上，你必须创建一个服务*而不是守护进程。无论如何，最可移植的解决方案可能是使用Celery或Twisted。 - pacha

Answer 2

我会使用消息代理，例如RabbitMQ或ActiveMQ。Flask进程将作业添加到消息队列中，长时间运行的工作进程（或进程池或工作进程）将从队列中取出作业以完成它们。工作进程可以更新数据库以允许Flask服务器知道作业的当前状态，并将此信息传递给客户端。

使用Celery似乎是一个不错的选择。

- Stephen Paulger

---

你能否分享一个创建长时间运行的工作进程来从队列中读取的示例？ - unnik