问题标签 [django-celery]

我正在尝试使用 Celery 实现以下场景：两个（相同）长时间运行的任务队列，一个用于“正常”，另一个用于“空闲”优先级。

我会让工作人员监控两个队列并首先从“正常”优先级队列中获取任务，如果它是空的，它将从“空闲”优先级中获​​取任务。

我的问题是：是否可以保证工作人员检查任务队列的顺序？此外，这是实施优先事项的正确方法吗？

背景：任务正在运行 ffmpeg 转码作业。“正常”优先级是新视频进来（必须尽快转码），“空闲”优先级是将旧（40.000+ 视频）存档重新转码为更新格式设置的任务。我没有多台服务器可用于进行多服务器任务分派。

正如我在top实用程序过程中看到的celery那样，消耗了大量的 CPU 时间。所以我想分析它。

我可以像这样在开发人员机器上手动完成：

但是要获得准确的时间，我需要在生产机器上对其进行分析。在该机器（Fedora 14）上，celery 由 init 脚本启动。例如

我已经弄清楚这些脚本最终最终会调用manage.py celeryd_multi。所以我的问题是如何告诉celeryd_multi在启用分析的情况下启动 celery？在我的情况下，这意味着将-m cProfile -o out.prof选项添加到python.

任何帮助深表感谢。

我想加快我的 django 视图处理速度，在视图中我必须进行几个 Web 服务调用（在我的情况下是 facebook graph api），这确实需要相当长的时间来处理（渲染视图大约需要 15-16 秒其中大部分用于从 facebook 获取数据）。所以而不是

我想这样做：

这样我就可以继续对朋友数据的处理，而无需等待检索个人资料信息。但是如果 celery worker(s) 很忙，那么它可能不会立即获取配置文件数据，因此视图的渲染可能比以前的方法花费更多甚至更多的时间。

第三种方法可能是与方法 2 中的所有操作相同，但如果任务尚未启动，则取消该任务并进行常规函数调用，而不是启动任务以获取配置文件信息。

以防万一有人建议使用 facebook 批处理请求获取个人资料和朋友信息：在我的情况下这是不可能的，因为上面的代码只是一个片段。当用户第一次访问我的应用程序时，我实际上是在中间件中获取配置文件。

我不确定以上 3 种方法中哪种方法更好。请建议是否有其他方法可以并行化 Web 请求。

当我尝试重试失败的任务时，我间歇性地（大约 20% 的时间）从 Celery 收到 IOError 异常。

这是我的任务：

例外：

RabbitMQ 日志

任何想法为什么会发生这种情况？

谢谢！

我正在使用Celery 对我制作的 CGI 应用程序中的作业进行排队。按照我的设置方式，Celery 通过设置CELERYD_CONCURRENCY = 1or使每个作业一次运行一个或两个= 2（因此它们不会拥挤处理器或因内存消耗而崩溃）。多亏了我在 StackOverflow 上得到的建议，队列工作得很好。

这些作业中的每一个都需要相当长的时间（大约 30 分钟串行），但具有令人尴尬的并行性。出于这个原因，我Pool.map习惯于拆分它并并行完成工作。它在命令行中运行良好，我使用新的多核芯片获得了大约 5 分钟的运行时间。

不幸的是，有一些限制不允许守护进程拥有子进程，当我在 CGI 队列中运行花哨的并行化代码时，我收到此错误：

AssertionError: 守护进程不允许有子进程

我注意到其他人也有类似的问题，但我找不到不需要完全放弃Pool.map并制作更复杂的线程代码的答案。

这里合适的设计选择是什么？我可以使用我的 Celery 队列轻松运行我的串行作业。我还可以在没有队列的情况下运行我更快的并行化作业。我应该如何解决这个问题，是否可以获得我想要的（队列和每个作业并行化）？

我有几个想法（有些很老套）：

• 发送到 Celery 队列的作业只是调用命令行程序。该程序可以随意使用 Pool，然后将结果数字和数据保存到文件中（就像现在一样）。
  缺点：我将无法检查作业的状态或查看它是否成功终止。此外，来自 CGI 的系统调用可能会导致安全问题。
• 显然，如果队列中的作业非常多，我可以充分利用 CPU 资源（通过设置 CELERYD_CONCURRENCY = 6 左右）；这将允许许多人同时“排在队列的最前面”。
  缺点：每个工作都会在队列最前面花费大量时间；如果队列未满，则不会有加速。此外，许多部分完成的作业将同时存储在内存中，使用更多的 RAM。
• 使用 Celery 的 @task 在子作业中并行化。然后，我不会设置 CELERYD_CONCURRENCY = 1，而是将其设置为 6（或者我希望一次允许在内存中的许多子作业）。
  缺点：首先，我不确定这是否会成功避免“任务中的任务”问题。但是，队列位置的概念可能会丢失，许多部分完成的作业可能会立即进入内存。
• 也许有一种方法可以调用 Pool.map 并指定线程是非守护线程的？或者也许我可以使用更轻量级的东西来代替 Pool.map？这类似于对另一个开放的 StackOverflow 问题采取的方法。另外，我应该注意，我通过 Pool.map 利用的并行化类似于线性代数，并且没有进程间通信（每个都独立运行并返回其结果而不与其他人交谈）。
• 扔掉 Celery 并使用 multiprocessing.Queue。那么也许有一些方法可以为我使用的每个线程使用相同的“线程深度”（即，也许所有线程都可以使用同一个池，避免嵌套）？

提前非常感谢。

我正在使用芹菜、rabbitmq 的项目工作。我想有权控制将队列推送任务到worker（celeryd）的间隔。

我有 task:recursive_task 它将安排相同的任务在 5 秒后执行，但如果由于某种原因该任务崩溃，则需要再次重新运行。我几乎抓住了每一个场景，但你永远不知道未来会发生什么。

我首先做了一个重复的任务：manage_tasks，它检查recursive_task的状态，会检查它是否长时间没有运行，是否成功完成，但这感觉不对。那么你将如何解决这个问题呢？

我们正在使用 celery 在 Windows 系统上进行一些 django 任务调度。几个月来一直很好，没有问题。然而今天，在重新启动后（这些几乎每天都发生）我无法让芹菜运行。Celerybeat 正常启动，但是当我运行 Celeryd 时，它开始加载，然后返回到 windows 命令行而没有抛出任何错误。

错误会使诊断变得更加容易，但没有它们我不知道从哪里开始。我对 Celery 几乎一无所知，这个负责一切的人早就离开了公司，所以我不知道该怎么办。

任何诊断问题的帮助都将受到欢迎！如果有任何信息需要我提供，请告诉我。

编辑

我找到了一个日志文件，其中包含以下内容：

谢谢！

我是 celery 的新手，可能做错了什么，但我已经花了很多时间试图弄清楚如何正确配置 celery。

因此，在我的环境中，我有 2 台远程服务器；一个是主要的（它有公共 IP 地址和数据库服务器、rabbitmq 服务器和运行我的 Web 应用程序的 Web 服务器等大部分东西都在那里），另一个用于我想使用 celery 从主服务器异步调用的特定任务。

我打算使用 RabbitMQ 作为代理和结果后端。Celery 配置非常基本：

当我在主服务器上运行一个worker时，任务执行得很好，但是当我在远程服务器上运行它时，只执行了几个任务，然后worker卡住了，无法执行任何任务。当我重新启动工作人员时，它会执行更多任务并再次卡住。任务内部没有什么特别之处，我什至尝试了一个只添加 2 个数字的测试任务。我尝试以不同的方式运行worker（妖魔化而不是妖魔化，设置不同的并发性并使用celeryd_multi），没有任何帮助。

可能是什么原因？我错过了什么？除了代理（RabbitMQ）之外，我是否必须在主服务器上运行某些东西？或者它是芹菜中的一个错误（我尝试了几个版本：2.2.4、2.3.3 和 dev，但都没有工作）？

嗯...我刚刚在本地工作人员上重现了同样的问题，所以我真的不知道它是什么......是否需要在每执行 N 个任务后重新启动 celery 工作人员？

任何帮助将不胜感激:)

在 django shell 中运行此脚本：

给出这个错误：

奇怪的是 chorus.processors.topics.tasks.test在我运行时肯定包含在 [Tasks] 打印输出中

为什么我会收到错误消息？