python - 使用 Python、Flask 和 Celery 的并发异步进程

Question

我正在开发一个小型但计算密集型的 Python 应用程序。计算密集型工作可以分解成几个可以同时执行的部分。我正在尝试确定一个合适的堆栈来完成此任务。

目前我计划在 Apache2+WSGI 上使用 Flask 应用程序和 Celery 作为任务队列。

在下面，如果有 3 个或更多 worker 可用，a_long_process()、another_long_process() 和 yet_another_long_process() 是否会同时执行？ Flask 应用程序会在进程执行时被阻塞吗？

来自 Flask 应用:

@myapp.route('/foo')
def bar():
    task_1 = a_long_process.delay(x, y)
    task_1_result = task_1.get(timeout=1)
    task_2 = another_long_process.delay(x, y)
    task_2_result = task_2.get(timeout=1)
    task_3 = yet_another_long_process.delay(x, y)
    task_3_result = task_3.get(timeout=1)
    return task_1 + task_2 + task_3

任务.py:

from celery import Celery
celery = Celery('tasks', broker="amqp://guest@localhost//", backend="amqp://")
@celery.task
def a_long_process(x, y):
    return something
@celery.task
def another_long_process(x, y):
    return something_else
@celery.task
def yet_another_long_process(x, y):
    return a_third_thing

Answer 1

您应该更改您的代码，以便工作人员可以并行工作:

@myapp.route('/foo')
def bar():
    # start tasks
    task_1 = a_long_process.delay(x, y)
    task_2 = another_long_process.delay(x, y)
    task_3 = yet_another_long_process.delay(x, y)
    # fetch results
    try:
        task_1_result = task_1.get(timeout=1)
        task_2_result = task_2.get(timeout=1)
        task_3_result = task_3.get(timeout=1)
    except TimeoutError:
        # Handle this or don't specify a timeout.
        raise
    # combine results
    return task_1 + task_2 + task_3

此代码将阻塞，直到所有结果可用(或达到超时)。

Will the Flask app be blocked while the processes are executing?

此代码只会阻止您的 WSGI 容器的一名工作人员。整个站点是否无响应取决于您使用的 WSGI 容器。 (例如 Apache + mod_wsgi、uWSGI、gunicorn 等)大多数 WSGI 容器会生成多个 worker，因此在您的代码等待任务结果时只有一个 worker 会被阻塞。

对于这种应用，我建议使用 gevent它为每个请求生成一个单独的 greenlet，并且非常轻量级。