我有 celery beat 和 celery（四名工人）批量做一些加工步骤。其中一项任务大致类似于“对于每个尚未创建 Y 的 X，创建一个 Y”。
该任务以半快速（10 秒）的速度定期运行。任务完成得非常快。还有其他任务正在进行。
我多次遇到过节拍任务明显积压的问题，因此同时执行相同的任务（来自不同的节拍时间），导致错误地重复工作。似乎任务是乱序执行的。
是否可以限制 celery beat 以确保一次只有一个未完成的任务实例？在任务上设置类似
rate_limit=5的“正确”方法吗？是否可以确保按顺序执行 beat 任务，例如，不是分派任务，而是 beat 将其添加到任务链中？
除了使这些任务本身以原子方式执行并且可以安全地同时执行之外，最好的处理方法是什么？这不是我所期望的击败任务的限制……</p>
任务本身的定义很天真：
@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.0197986490093s 内成功：无[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 添加 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 内成功：无
我们目前使用 Celery 3.1.4 和 RabbitMQ 作为传输。
编辑丹，这是我想出的：
丹，这就是我最终使用的：
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