我正在使用一个面向软件的架构，它有多个celeryworker(我们称它们为worker1、worker2和worker3)。所有三个工作人员都是独立的实体(即，独立的代码库、独立的repos、独立的celery实例、独立的机器)，它们都没有连接到Django应用程序。通过基于Django、MySQL支持的RESTfulAPI与这三个工作人员中的每一个进行通信。在开发中，这些服务都在一个vagrantbox上，每一个都作为一个独立的机器运行在一个单独的端口上。我们有一个RabbitMQ代理来处理所有Celery任务。通过这些服务的典型路径可能如下所示:worker1从设备获取消息，进行
我已经调用了一个使用urllib2远程获取一些信息的任务数千次。这些任务是用随机eta安排的(一周内)，因此它们不会同时访问服务器。有时我会收到404，有时不会。我正在处理错误以防万一。在RabbitMQ控制台中，我可以看到16条未确认的消息:我停止了celery，清除了队列并重新启动了它。16条未确认的消息仍然存在。我还有其他任务进入同一个队列，但也没有执行。清除后，我尝试提交另一个任务，它的状态仍然是就绪:有什么办法可以找出消息未被确认的原因吗？版本:celery==3.1.4{rabbit,"RabbitMQ","3.5.3"}celeryapp.pyCELERYBEAT_SCH
我的Celery任务引发自定义异常NonTransientProcessingError，然后被AsyncResult.get()捕获.任务.py:classNonTransientProcessingError(Exception):pass@shared_task()defthrow_exception():raiseNonTransientProcessingError('ErrorraisedbyPOCmodelfortestpurposes')在Python控制台中:frommy_app.tasksimport*r=throw_exception.apply_async()t
我有一个Flask应用程序，注册如下:APP=Flask(__name__)APP.config.from_object('config')我已经为URL定义了一个View，其中调用了一个与数据库交互的函数。fromtasksimportsome_func...some_func.delay(params)在tasks.py文件中，我正在创建一个Celery实例，如下所示:#Configvaluesdon'tworkherecelery=Celery('tasks',broker='amqp://',backend='amqp://')...@celery.task()defsome_
所以我将django与celery一起使用。rabbitmq是经纪人。redis是缓存。mysql是数据库。(本地主机中的所有内容)我正在使用python2.7并使用基于virtualenv的虚拟环境我在默认端口启动redis服务器(本地)在一个新的终端中，我运行pythonmanage.pyrunserver在一个新终端中，我像这样启动celerycelery-Ads_djangoworker-B-lwarning这个程序曾经有效。但是现在当我运行celery命令时，我得到以下内容[2016-07-1209:15:20,113:CRITICAL/MainProcess]Frequen
出于某种原因，每当我在Celery中创建并运行新任务时，都会出现返回结果的问题。第一个任务完美返回，但对于所有后续任务，结果始终未决。我检查了Celery日志，它得到了正确的结果，没有错误，但就是无法返回。如果有帮助，我正在运行rabbitmq作为我的后端。 最佳答案 好吧，事实证明我只需要明确指定一个后端。添加:CELERY_RESULT_BACKEND="amqp"我的设置文件似乎解决了所有问题。 关于python-celery不返回结果，我们在StackOverflow上找到一个类
我曾使用10个线程为我的项目运行celery。它工作得很好。现在，当我使用命令时:celery-Atasksworker--loglevel=info--concurrency10celery卡在:[2014-08-0712:55:21,013:INFO/MainProcess]Connectedtoamqp://guest:**@127.0.0.1:5672//[2014-08-0712:55:21,025:INFO/MainProcess]mingle:searchingforneighbors早些时候它显示就绪状态。当我尝试使用ctrl+c结束它时，它没有关闭。[2014-08-
Celery会发送任务给空闲的worker。我有一个任务每5秒运行一次，我希望这个任务只发送给一个指定的工作人员。其他任务可以共享剩下的workercelery可以这样做吗？？我想知道这个参数是什么:CELERY_TASK_RESULT_EXPIRES是不是表示任务不会发给队列中的某个worker？或者如果运行时间太长它会停止任务吗？ 最佳答案 当然可以。最好的方法是使用不同的队列将celeryworker分开。您只需要确保您需要的任务进入单独的队列，并且您的工作人员正在监听特定的队列。长话短说:http://docs.celery
如何限制可以同时运行的特定Celery任务的实例数？我有一个处理大文件的任务。我遇到了一个问题，用户可能会启动多个任务，导致服务器在尝试一次处理太多文件时耗尽CPU和内存。我想确保在任何给定时间只运行这种类型任务的N个实例，并且其他任务将在调度程序中排队，直到其他任务完成。我看到有一个rate_limit任务装饰器中的选项，但我认为这不是我想要的。如果我正确理解文档，这只会限制任务启动的速度，但不会限制运行的任务总数，所以这会使我的服务器崩溃得更慢......但它仍然会尽管如此还是崩溃了。 最佳答案 您必须设置额外的队列并为其设置所
我想在celery中的工作节点之间共享小块信息(例如缓存的授权token、统计信息等)。如果我在我的任务文件中创建一个全局变量，它对每个工作人员都是唯一的(我的工作人员是进程，并且有1个任务/执行的生命周期)。最佳做法是什么？我是否应该在外部保存状态(DB)，创建一个老式的共享内存(由于celery中不同的池实现可能很困难)？提前致谢! 最佳答案 我终于找到了一个像样的解决方案——corepythonmultiprocessing-Manager:frommultiprocessingimportManagermanag=Manag
