标签: django-celery

我正在使用Celery 2.4.6和django-celery 2.4.2.

当我根据这个问题的分辨率配置Celery使用Amazon SQS时:使用Amazon SQS的Celery

我在Django管理员的celerycam表中看不到任何内容.如果我切换回RabbitMQ,任务会再次开始显示.

我有很多(现在的40+)的SQS命名是这样的队列:"celeryev-92e068c4-9390-4c97-bc1d-13fd6e309e19",它看起来像他们可能与(一些上了年纪的人,甚至在他们的活动),但没有任何东西出现在数据库中,我看到celerycam日志中没有错误.

关于问题可能是什么或如何进一步调试的任何建议将非常感激.

我用Django ORM设置Celery作为后端.试图监控幕后发生的事情.

• 我用-E标志 开始了celerydpython manage.py celeryd -E -l INFO -v 1 -f /path/to/celeryd.log

• 启动celerycam,默认快照频率为1秒.python mannage.py celerycam

我可以看到正在执行的任务(在芹菜日志中)和存储的结果(数据模型会定期被这些任务更改).但是,Django管理面板中的任务/工作人员页面显示零项.当我开始celeryev(显示o events/tasks/workers)时,同样的事情发生了.

是否需要更改其他配置设置才能实现监控？

我的软件堆栈:Django 1.1,Celery 2.2.4,Python 2.4

我一直在与Django/Celery文档争论一段时间,需要一些帮助.

我希望能够使用django-celery运行Periodic Tasks.我已经在互联网(和文档)周围看到了几种不同的格式和模式,以了解如何使用Celery实现这一目标...

有人可以帮助创建,注册和执行django-celery定期任务的基本功能示例吗？特别是,我想知道是否应该编写一个扩展PeriodicTask类并注册它的任务,或者我是否应该使用@periodic_task装饰器,或者我是否应该使用@task装饰器然后为任务设置一个时间表执行.

我不介意这三种方式是否可行,但我希望看到一个至少有一种方法可行的例子.非常感谢您的帮助.

当我有类似以下的东西

group1 = group(task1.si(), task1.si(), task1.si())
group2 = group(task2.si(), task2.si(), task2.si())

workflow = chain(group1, group2, task3.si())

直观的解释是task3应该只在第2组中的所有任务完成后执行.

实际上,任务3在group1启动但尚未完成时执行.

我究竟做错了什么？

我已经设置芹菜使用他们的daemonization指令来处理我的django应用程序(http://docs.celeryproject.org/en/latest/tutorials/daemonizing.html#daemonizing)

这是我的测试任务

@periodic_task(run_every=timedelta(seconds=10))
def debugger():
    logger.info("Running debugger")
    raise Exception('Failed')

我需要一种方法来了解此任务(调试器)由于异常而失败.Celery的日志文件打印logger.info("运行调试器")日志,但它不记录异常.我错过了什么,或者我应该以其他方式找到失败的任务？

我会将多个任务添加到芹菜队列并等待结果.我有各种想法如何利用某种形式的共享存储(memcached,redis,db等)实现这一点,但是,我会认为这是Celery可以自动处理但我无法在线找到任何资源的东西.

代码示例

def do_tasks(b):
    for a in b:
        c.delay(a)

    return c.all_results_some_how()

我试图异步使用Web服务,因为它需要45秒才能返回.不幸的是,这个Web服务也有些不可靠,可能会出错.我已经设置django-celery并执行了我的任务,这些工作正常,直到任务失败max_retries.

这是我到目前为止:

@task(default_retry_delay=5, max_retries=10)
def request(xml):
    try:
        server = Client('https://www.whatever.net/RealTimeService.asmx?wsdl')
        xml = server.service.RunRealTimeXML(
            username=settings.WS_USERNAME,
            password=settings.WS_PASSWORD,
            xml=xml
        )
    except Exception, e:
        result = Result(celery_id=request.request.id, details=e.reason, status="i")
        result.save()
        try:
            return request.retry(exc=e)
        except MaxRetriesExceededError, e:
            result = Result(celery_id=request.request.id, details="Max Retries Exceeded", status="f")
            result.save()
            raise
    result = Result(celery_id=request.request.id, details=xml, status="s")
    result.save()
    return result

不幸的是,MaxRetriesExceededError没有被抛出retry(),所以我不知道如何处理这项任务的失败.Django已经将HTML返回给客户端,我正在检查Result通过AJAX 的内容,它永远不会达到完全失败f状态.

所以问题是:当Celery任务超过时,如何更新我的数据库max_retries？

我刚刚开始使用django-celery,我想将celeryd设置为守护进程.但是,这些说明似乎表明它一次只能配置一个站点/项目.芹菜可以处理多个项目,还是只能处理一个项目？而且,如果是这种情况,是否有一种干净的方法来设置celeryd为每个配置自动启动,这需要我为每个配置创建一个单独的init脚本？

我有一个check_orders任务,定期执行.它创建了一组任务,以便我可以计算执行任务所花费的时间,并在完成任务时执行某些操作(这是res.join [1]和grouped_subs的目的)分组的任务是成对的链式任务.

我想要的是当第一个任务不满足条件(失败)时不执行链中的第二个任务.对于我的生活,我无法理解这一点,我觉得这对于作业队列管理器来说是非常基本的功能.当我尝试在[2]之后注释掉的东西(引发异常,删除回调)......由于某种原因,我们在check_orders中遇到了join()(它打破了组).我已经尝试将ignore_result设置为False以及所有这些任务,但它仍然不起作用.

@task(ignore_result=True)
def check_orders():
    # check all the orders and send out appropriate notifications
    grouped_subs = []

    for thingy in things:
       ...

        grouped_subs.append(chain(is_room_open.subtask((args_sub_1, )), 
                        notify.subtask((args_sub_2, ), immutable=True)))

    res = group(grouped_subs).apply_async()

    res.join()         #[1]
    logger.info('Done checking orders at %s' % current_task.request.id))

@task(ignore_result=True)
def is_room_open(args_sub_1):
    #something time consuming
    if http_req_and_parse(args_sub_1):
        # go on and do the notify task
        return True
    else:
        # [2]
        # STOP THE CHAIN SOMEHOW! Don't execute the rest of the chain, how?
        # None of the following …

如果我在芹菜运行时对tasks.py进行了更改,是否有一种机制可以重新加载更新的代码？或者我必须关闭芹菜重新装载？

我读过芹菜--autoreload在旧版本中有一个参数,但我在当前版本中找不到它:

celery: error: unrecognized arguments: --autoreload