我计划使用Celery来处理由我的主服务器事件触发的推送通知和电子邮件的发送。这些任务需要打开与外部服务器(GCM、APS、电子邮件服务器等)的连接。它们可以一次处理一个，也可以通过单个连接批量处理以获得更好的性能。通常会在短时间内分别触发这些任务的多个实例。例如，在一分钟内，可能有几十个推送通知需要发送给具有不同消息的不同用户。在Celery中处理这个问题的最佳方法是什么？似乎天真的方法是简单地为每条消息分配不同的任务，但这需要为每个实例打开一个连接。我希望有某种任务聚合器允许我处理，例如'所有未完成的推送通知任务'。有这样的东西吗？有没有更好的方法来解决这个问题，例如附加到事件任务
我正在使用此命令运行我的服务器:celeryworker-Qq1,q2-c2这表明我的服务器将处理队列q1和q2上的所有任务，并且我有2个worker正在运行。我的服务器应该支持2个不同的任务:@celery.task(name='test1')deftest1():print"test1"time.sleep(3)@celery.task(name='test2')deftest2():print"test2"如果我将我的test1任务发送到q1队列并将test2任务发送到q2，两个工作人员都将同时运行任务。所以结果将是:test1test2test1test2...现在我需要的是我
我需要允许用户提交对非常非常大的作业的请求。我们说的是100GB的内存和20小时的计算时间。这让我们公司花了很多钱，所以规定任何时候只能运行2个作业，已经有2个作业再请求新的作业会被拒绝(并通知用户服务器忙)。我当前的解决方案使用concurrent.futures中的执行器，并且需要将Apache服务器设置为仅运行一个进程，从而降低响应速度(当前用户数非常少，所以暂时没问题)。如果可能的话，我想为此使用Celery，但我没有在文档中看到任何方式来完成此特定设置。如何在Django应用程序的后台运行有限数量的作业，并在作业因服务器繁忙而被拒绝时通知用户？ 最
我正在尝试设置每分钟都运行的两个任务。有什么办法可以将它们组合在一起运行吗？我在celeryconfig.py中指定了CELERYBEAT_SCHEDULE，如下所示:CELERYBEAT_SCHEDULE={'every-minute':{'task':'tasks.add','schedule':crontab(minute='*/1'),'args':(1,2)},}所以如果我想运行两个任务，我会期待这样的事情吗？CELERYBEAT_SCHEDULE={'every-minute':{'task':['tasks.add','task.multiply'],'schedule'
我正在运行Django1.8+Celery4.0.2Celery配置良好，可以在redis后端本地运行我的Django任务。但是当我尝试使用CELERY_ALWAYS_EAGER=True设置时，此设置无效。其他设置不是这种情况，例如CELERY_TIMEZONE具体来说，在pdb中，我看到app.conf.task_always_eager是Falselib/python2.7/site-packages/celery/app/task.py(520)apply_async()SosomehowCELERY_ALWAYS_EAGERisnotpickedupandhasnoeffec
我正在尝试使用celery在一组服务器上安排和运行任务。每个任务运行时间都比较长(几个小时)，并且涉及使用子进程调用具有给定输入的特定程序。该程序在stdout和stderr中产生大量输出。有没有办法近乎实时地向客户端显示程序产生的输出？流式传输输出，以便客户端无需登录服务器即可观看服务器上运行的任务产生的输出？ 最佳答案 您没有指定很多要求和约束。我假设您已经在某处拥有一个Redis实例。你可以做的是逐行读取其他进程的输出并通过redis发布:这是一个示例，您可以在其中将数据echo到文件/tmp/foo中以进行测试:import
这可能是一个愚蠢的问题，但它让我从Ruby背景中难过。当我尝试打印时，我有一个看起来像这样的对象。printcelery.AsyncResult.task_id>>>我原以为这里会打印task_id属性的实际值。如何获得实际值？更新1@celery.taskdefscan(host):printcelery.AsyncResult.task_idcmd='ps-ef'cm=shlex.split(cmd)scan=subprocess.check_output(cm)returnscan最好的问候。 最佳答案 短篇小说，在函数sca
乍一看，我非常喜欢Celery中的“批处理”功能，因为我需要在调用API之前对一定数量的ID进行分组(否则我可能会被踢出局)。不幸的是，在进行一些测试时，批处理任务似乎不能很好地与其余的Canvas基元(在本例中为链)配合使用。例如:@a.task(base=Batches,flush_every=10,flush_interval=5)defget_price(requests):forrequestinrequests:a.backend.mark_as_done(request.id,42,request=request)print"filter_by_price"+str([r
我正在尝试使用Celery作为Twisted应用程序的控制channel。我的Twisted应用程序是一个抽象层，它为各种本地运行的进程(通过ProcessProtocol)提供标准接口(interface)。我想使用Celery来远程控制它——AMQP似乎是从中央位置控制许多Twisted应用程序的理想方法，我想利用Celery基于任务的功能，例如任务重试、子任务等这并没有像我计划的那样工作，我希望有人能帮助我指明正确的方向以实现这一目标。我在运行脚本时试图实现的行为是:开始一个稍微修改过的celeryd(见下)等待Celery任务收到“启动流程”任务时，生成ProcessProto
我关注了celerydocs在我的开发机器上定义2个队列。我的celery设置:CELERY_ALWAYS_EAGER=TrueCELERY_TASK_RESULT_EXPIRES=60#1minsCELERYD_CONCURRENCY=2CELERYD_MAX_TASKS_PER_CHILD=4CELERYD_PREFETCH_MULTIPLIER=1CELERY_CREATE_MISSING_QUEUES=TrueCELERY_QUEUES=(Queue('default',Exchange('default'),routing_key='default'),Queue('feed
