Celery 4.4.0 - 使用 SQS 作为代理时禁用预取

Question

我已经查看了多个关于此的 SO 问题，但仍然没有找到解决方案，而且我已经为此工作了几天，因此我们将不胜感激！

我使用 celery 4.4.0、SQS 作为代理和 Django 2.2，我的大部分任务都很长（1 到 4 小时）。
这是启动工人的命令：

celery worker -A config.celeryconfig:app -Ofair --prefetch-multiplier=1 --max-tasks-per-child=2

我的 Django 配置文件中的这个配置：

CELERYD_PREFETCH_MULTIPLIER = 1
CELERY_ACKS_LATE = True
task_acks_late = True # I wasn't sure what's the name of the ack late configuration.

BROKER_TRANSPORT_OPTIONS = {
    'polling_interval': 3,
    'region': 'us-east-1',
    'visibility_timeout': 3600 # 1 hour,
}

场景 - 假设我们有 2 名工人 - 1,2 名工人，每个人都有一个子流程和两个任务 a（超过 1 小时），b，c，我几乎在同一时间派遣了他们：
工人 1 接到任务 a
工人 1 拿起了任务 c（但没有执行）
工人 2 接了任务 b
工人 2 完成了任务 b
工人 2 接了任务 c（因为 visibility_timeout）
工人 2 完成了任务 c
工人 1 完成了任务 a
工人 1 从任务 c 开始
工人 1 完成任务 c

所以：

1. 工作人员仍在预取任务。
2. （更糟）同一个任务可以执行两次（甚至更多，如果有 更多的工人和执行时间比visibility_timeout) 长。在这里https://github.com/celery/celery/issues/4400有详细讨论。

我已经意识到上述问题的解决方案是禁用预取行为，但到目前为止我无法实现。

我很沮丧，所以如果您对如何解决这个问题有任何想法 - 请告诉我！

几点注意事项：
*我在我的生产环境中也看到了它，每个工人有超过 1 个子流程。
*我曾经使用倒计时（导致 ETA 任务）分派任务，但已禁用它并且问题仍然存在。

• 我没有使用 result_backend，我正在处理我创建的数据库中模型上的所有内容。
  *我不想设置CONCURRENCY=1，因为我希望每台机器有一个工作人员，机器中有 subprocess=CPU 内核。
  *我已经尝试了很多上述芹菜配置的组合，但没有一个有效（上面列出的一个基于此评论 - https://stackoverflow.com/a/58958823）
  *我不喜欢使用的另一种可能的解决方案是将 visibility_timeout 增加到一个非常大的数字。但这可能导致一个任务一个接一个地提交所有长任务，并且没有在所有工作人员之间分配它们。
  *不确定是否相关 - 我在 EC2 机器上使用 ElasticBeanstalk 部署 celery。 *我目前正在考虑的另一种可能的解决方案 - 正在检查任务的状态（我们使用 Pending/In Progress 等状态），并且仅当它处于未决状态时 - 继续（这可能无法解决整个用例，因为可能竞争条件，但它应该可以解决大部分问题）。

Answer 1

好吧，我从 Celery 文档中了解到，“CELERYD_PREFETCH_MULTIPLIER”只会改变一次预取一项任务的行为，但它不会限制预取功能。因此，如果您的并发 > 1，它仍然会为每个工作人员预取 1 个任务。尝试设置 --concurrency=1

一次预取多少消息乘以并发进程数。默认值为 4（每个进程 4 条消息）。