用芹菜执行“独特”任务

Question

我用芹菜更新我的新闻聚合网站的RSS订阅源。我为每个feed使用一个@task，事情看起来运行得很好。

但有一个细节我不确定是否能很好地处理：所有feed每分钟都会更新一次@periodic_task，但是如果一个feed在启动新的feed时仍在从上一个periodic task更新呢？（例如，如果feed非常慢，或者脱机，并且任务处于重试循环中）

当前我存储任务结果并检查其状态，如下所示：

import socket
from datetime import timedelta
from celery.decorators import task, periodic_task
from aggregator.models import Feed


_results = {}


@periodic_task(run_every=timedelta(minutes=1))
def fetch_articles():
    for feed in Feed.objects.all():
        if feed.pk in _results:
            if not _results[feed.pk].ready():
                # The task is not finished yet
                continue
        _results[feed.pk] = update_feed.delay(feed)


@task()
def update_feed(feed):
    try:
        feed.fetch_articles()
    except socket.error, exc:
        update_feed.retry(args=[feed], exc=exc)

也许有一种更复杂/更稳健的方法，可以使用我错过的芹菜机制来达到同样的效果？

Answer 1

根据MattH的回答，您可以使用这样的装饰：

def single_instance_task(timeout):
    def task_exc(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            lock_id = "celery-single-instance-" + func.__name__
            acquire_lock = lambda: cache.add(lock_id, "true", timeout)
            release_lock = lambda: cache.delete(lock_id)
            if acquire_lock():
                try:
                    func(*args, **kwargs)
                finally:
                    release_lock()
        return wrapper
    return task_exc

然后，像这样使用它。。。

@periodic_task(run_every=timedelta(minutes=1))
@single_instance_task(60*10)
def fetch_articles()
    yada yada...

Answer 2

来自官方文件：Ensuring a task is only executed one at a time。

Answer 3

使用https://pypi.python.org/pypi/celery_once似乎做得很好，包括报告错误和针对某些参数进行唯一性测试。

你可以这样做：

from celery_once import QueueOnce
from myapp.celery import app
from time import sleep

@app.task(base=QueueOnce, once=dict(keys=('customer_id',)))
def start_billing(customer_id, year, month):
    sleep(30)
    return "Done!"

只需要在项目中进行以下设置：

ONCE_REDIS_URL = 'redis://localhost:6379/0'
ONCE_DEFAULT_TIMEOUT = 60 * 60  # remove lock after 1 hour in case it was stale