Celery Beat：一次限制为单个任务实例

Question

我有芹菜和芹菜（四名工人）批量做一些加工步骤。其中一项任务大致是这样的，“对于每个没有创建Y的X，创建Y.”

该任务以半快速（10秒）定期运行。任务很快完成。还有其他任务正在进行中。

我多次遇到这个问题，其中节拍任务显然已经积压，因此同时执行相同的任务（来自不同的节拍时间），导致错误的重复工作。似乎任务也是无序执行的。

是否有可能限制芹菜节拍以确保一次只有一个突出的任务实例？设置像rate_limit=5这样的任务是“正确”的方式吗？
是否有可能确保按顺序执行节拍任务，例如，而不是派遣任务，节拍将其添加到任务链？
处理这个问题的最佳方法是什么，除了使这些任务本身以原子方式执行并且可以安全地同时执行？这不是我对节拍任务的预期限制......

任务本身是天真地定义的：

@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] TaskPool：应用＃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.0197986490093s：无
[00:26.806] TaskPool：应用＃1
[00:26.836] TaskPool：应用＃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添加Y＃9725
[01:30.070] foocorp.tasks.add_y_to_xs [＃3]：为X id添加Y＃9725
[01:30.074]任务foocorp.tasks.add_y_to_xs [＃1]成功完成0.0594762689434s：无
[01:30.087]任务foocorp.tasks.add_y_to_xs [＃3]成功进入0.0352867960464s：无

我们目前正在使用带有RabbitMQ的Celery 3.1.4作为传输。

编辑丹，这是我想出的：

丹，这是我最终使用的：

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

和芹菜任务装饰器（仅用于定期任务）：

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

另一答案

Answer 2

另一答案

Answer 3

另一答案

Answer 4

另一答案