python - celery 节拍 : Limit to single task instance at a time

Question

我有 celery beat 和 celery(四个 worker )批量做一些加工步骤。其中一项任务大致是这样的:“对于每个尚未创建 Y 的 X，创建一个 Y。”

任务以半快速(10 秒)的速度定期运行。任务完成得非常快。还有其他任务正在进行中。

我已经多次遇到节拍任务明显积压的问题，因此同一任务(来自不同的节拍时间)同时执行，导致错误地重复工作。任务似乎也是乱序执行的。

1. 是否可以限制 celery beat 以确保一次只有一个未完成的任务实例？在任务上设置类似 rate_limit=5 的设置是否是执行此操作的“正确”方法？

2. 是否可以确保节拍任务按顺序执行，例如beat 不是分派(dispatch)任务，而是将其添加到任务链中？

3. 除了使这些任务本身以原子方式执行并且可以安全地并发执行之外，处理此问题的最佳方法是什么？这不是我所期望的节拍任务的限制……

任务本身的定义很简单:

@periodic_task(run_every=timedelta(seconds=10))
def add_y_to_xs():
    # Do things in a database
    return

这是一个实际的(清理过的)日志:

• [00:00.000] foocorp.tasks.add_y_to_xs 已发送。 id->#1
• [00:00.001] 收到任务:foocorp.tasks.add_y_to_xs[#1]
• [00:10.009] foocorp.tasks.add_y_to_xs 已发送。 id->#2
• [00:20.024] foocorp.tasks.add_y_to_xs 已发送。 id->#3
• [00:26.747] 收到任务:foocorp.tasks.add_y_to_xs[#2]
• [00:26.748] 任务池:应用 #2
• [00:26.752] 收到任务:foocorp.tasks.add_y_to_xs[#3]
• [00:26.769] 接受的任务:foocorp.tasks.add_y_to_xs[#2] pid:26528
• [00:26.775] 任务 foocorp.tasks.add_y_to_xs[#2] 在 0.0197986490093 秒内成功:无
• [00:26.806] 任务池:应用 #1
• [00:26.836] 任务池:应用 #3
• [01:30.020] 接受的任务:foocorp.tasks.add_y_to_xs[#1] pid:26526
• [01:30.053] 接受的任务:foocorp.tasks.add_y_to_xs[#3] pid:26529
• [01:30.055] foocorp.tasks.add_y_to_xs[#1]:为 X id #9725 添加 Y
• [01:30.070] foocorp.tasks.add_y_to_xs[#3]:为 X id #9725 添加 Y
• [01:30.074] 任务 foocorp.tasks.add_y_to_xs[#1] 在 0.0594762689434 秒内成功:无
• [01:30.087] 任务 foocorp.tasks.add_y_to_xs[#3] 在 0.0352867960464 秒内成功:无

我们目前使用 Celery 3.1.4 和 RabbitMQ 作为传输。

编辑 Dan，这是我的想法:

Dan，这是我最终使用的:

from sqlalchemy import func
from sqlalchemy.exc import DBAPIError
from contextlib import contextmanager


def _psql_advisory_lock_blocking(conn, lock_id, shared, timeout):
    lock_fn = (func.pg_advisory_xact_lock_shared
               if shared else
               func.pg_advisory_xact_lock)
    if timeout:
        conn.execute(text('SET statement_timeout TO :timeout'),
                     timeout=timeout)
    try:
        conn.execute(select([lock_fn(lock_id)]))
    except DBAPIError:
        return False
    return True


def _psql_advisory_lock_nonblocking(conn, lock_id, shared):
    lock_fn = (func.pg_try_advisory_xact_lock_shared
               if shared else
               func.pg_try_advisory_xact_lock)
    return conn.execute(select([lock_fn(lock_id)])).scalar()


class DatabaseLockFailed(Exception):
    pass


@contextmanager
def db_lock(engine, name, shared=False, block=True, timeout=None):
    """
    Context manager which acquires a PSQL advisory transaction lock with a
    specified name.
    """
    lock_id = hash(name)

    with engine.begin() as conn, conn.begin():
        if block:
            locked = _psql_advisory_lock_blocking(conn, lock_id, shared,
                                                  timeout)
        else:
            locked = _psql_advisory_lock_nonblocking(conn, lock_id, shared)
        if not locked:
            raise DatabaseLockFailed()
        yield

以及 celery 任务装饰器(仅用于周期性任务):

from functools import wraps
from preo.extensions import db


def locked(name=None, block=True, timeout='1s'):
    """
    Using a PostgreSQL advisory transaction lock, only runs this task if the
    lock is available. Otherwise logs a message and returns `None`.
    """
    def with_task(fn):
        lock_id = name or 'celery:{}.{}'.format(fn.__module__, fn.__name__)

        @wraps(fn)
        def f(*args, **kwargs):
            try:
                with db_lock(db.engine, name=lock_id, block=block,
                             timeout=timeout):
                    return fn(*args, **kwargs)
            except DatabaseLockFailed:
                logger.error('Failed to get lock.')
                return None
        return f
    return with_task

Answer 1

from functools import wraps
from celery import shared_task


def skip_if_running(f):
    task_name = f'{f.__module__}.{f.__name__}'

    @wraps(f)
    def wrapped(self, *args, **kwargs):
        workers = self.app.control.inspect().active()

        for worker, tasks in workers.items():
            for task in tasks:
                if (task_name == task['name'] and
                        tuple(args) == tuple(task['args']) and
                        kwargs == task['kwargs'] and
                        self.request.id != task['id']):
                    print(f'task {task_name} ({args}, {kwargs}) is running on {worker}, skipping')

                    return None

        return f(self, *args, **kwargs)

    return wrapped


@shared_task(bind=True)
@skip_if_running
def test_single_task(self):
    pass


test_single_task.delay()