带有 Amazon SQS 的 Celery答案

【问题标题】：Celery with Amazon SQS带有 Amazon SQS 的 Celery
【发布时间】：2011-12-24 07:34:03
【问题描述】：

我想使用Amazon SQS 作为Celery 支持的代理。 Kombu 的 SQS 传输实现是 Celery 所依赖的。但是没有足够的文档来使用它，所以我找不到如何在 Celery 上配置 SQS。有人在 Celery 上成功配置 SQS 吗？

【问题讨论】：

标签： amazon-web-services celery amazon-sqs kombu

【解决方案1】：

我多次遇到这个问题，但仍然不完全确定如何设置 Celery 以与 SQS 一起使用。事实证明，使用最新版本的 Kombu 和 Celery 非常容易。作为另一个答案中提到的 BROKER_URL 语法的替代方法，您可以像这样简单地设置传输、选项、用户和密码：

BROKER_TRANSPORT = 'sqs'
BROKER_TRANSPORT_OPTIONS = {
    'region': 'us-east-1',
}
BROKER_USER = AWS_ACCESS_KEY_ID
BROKER_PASSWORD = AWS_SECRET_ACCESS_KEY

这解决了 URL 解析器的一个所谓的问题，即不允许在您的 API 密钥中使用正斜杠，这在 AWS 中似乎很常见。由于似乎还没有关于该主题的大量信息，因此我还在这里写了一篇关于该主题的简短博客文章：

http://www.caktusgroup.com/blog/2011/12/19/using-django-and-celery-amazon-sqs/

【讨论】：

+1 感谢您为此 @tobias 所做的工作。好的博文和 cmets 中的讨论内容非常丰富。让我们更新！
亚马逊 SQS 是否变得更快了？我还没有使用它，主要是因为我听到很多关于可怕延迟（> 2 分钟）的报告，例如，在任务出现在队列中之前。
这对我不起作用。 Boto 的身份验证系统出现错误 - 没有处理程序准备好进行身份验证。检查了 1 个处理程序。 ['HmacAuthV4Handler'] 检查您的凭据
如果我们有多个 SQS 队列怎么办。我们如何告诉 python 使用哪个队列？

【解决方案2】：

我正在使用 Celery 3.0，并且在使用 BROKER_USER / BROKER_PASSWORD 设置启动工作程序时收到弃用警告。

我查看了 kombo.utils.url._parse_url 中的 SQS URL 解析，它在 URL 的用户名和密码元素上调用 urllib.unquote。

因此，为了解决带有正斜杠的密钥问题，我能够成功地将以下内容用于 BROKER_URL：

import urllib
BROKER_URL = 'sqs://%s:%s@' % (urllib.quote(AWS_ACCESS_KEY_ID, safe=''),
                               urllib.quote(AWS_SECRET_ACCESS_KEY, safe=''))

我不确定访问键中是否可以包含正斜杠，但引用它也没有什么坏处。

【讨论】：

这与我在 celery 3.1 上的密钥中的斜线一起使用
来自 celery 4.1.0 文档：The login credentials can also be set using the environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY, in that case the broker URL may only be sqs://. 我无法让它工作，但明确构建 BROKER_URL 因为这个答案确实很好用。
stackoverflow.com/questions/31827012/… 对于python 3

【解决方案3】：

对于遇到这个问题的任何人，我都能让 Celery 开箱即用地与 SQS 一起工作（无需修补），但我确实需要更新到最新版本的 Celery 和 Kombu 才能工作（截至目前为 1.4.5 和 1.5.1）。使用上面的配置行，它应该可以工作（尽管您可能想要更改默认区域）。

问题：为了使用上面的 URL 格式，您需要确保您的 AWS 密码不包含斜杠，因为这会混淆 URL 解析器。不断产生新的秘密，直到你得到一个没有斜线的秘密。

【讨论】：

一个答案应该作为一个独立的响应独立存在。它应该包含所有需要理解的信息。在其他地方引用其他内容（“上面的配置行”）没有帮助，尤其是考虑到页面上答案的位置是动态的。

【解决方案4】：

没有人回答这个问题。无论如何，我尝试使用 Amazon SQS 配置 Celery，并且似乎取得了小小的成功。

Kombu 应该为此打补丁，所以我写了一些补丁，还有my pull request。您可以通过在修补过的 Kombu 上的 Celery 中设置 BROKER_URL 的 sqs:// 方案来配置 Amazon SQS。例如：

BROKER_URL = 'sqs://AWS_ACCESS:AWS_SECRET@:80//'
BROKER_TRANSPORT_OPTIONS = {
    'region': 'ap-northeast-1',
    'sdb_persistence': False
}

【讨论】：

太棒了！尝试了你的补丁分支，SQS 终于工作了，但是工作人员一遍又一遍地完成相同的任务（甚至具有相同的 taskid！）。所以我回到使用rabbitmq作为经纪人。你在生产中使用 SQS+celery 吗？面临任何此类问题？
@sajal 我遇到了完全相同的问题，所以我终于回到使用 RabbitMQ。
@sajal：创建 SQS 队列时，您需要将默认可见性超时设置为大于您期望任务运行的最长时间的某个时间。这是 SQS 将消息传递给一个消费者后对所有其他消费者不可见的时间。我相信默认值为 30 秒。因此，如果一个任务花费的时间超过 30 秒，SQS 会将相同的消息传递给另一个消费者，因为它假定第一个消费者已经死亡并且没有完成任务。

【解决方案5】：

我在 IAM 控制台中重新生成了凭证，直到我得到一个没有斜杠的密钥 (/)。解析问题仅与该字符有关，因此如果您的秘密没有问题，您会没事的。

不是最优雅的解决方案，但绝对可以让代码保持干净。

【讨论】：

【解决方案6】：

Python 3 更新，从 AWS KEY 中删除反斜杠。

from urllib.parse import quote_plus
BROKER_URL = 'sqs://{}:{}@'.format(
    quote_plus(AWS_ACCESS_KEY_ID), 
    quote_plus(AWS_SECRET_ACCESS_KEY)
)

【讨论】：

【解决方案7】：

我能够使用 kombu 在 celery 4.3 (python 3.7) 上配置 SQS。

from kombu.utils.url import quote

CELERY_BROKER_URL = 'sqs://{AWS_ACCESS_KEY_ID}:{AWS_SECRET_ACCESS_KEY}@'.format(
    AWS_ACCESS_KEY_ID=quote(AWS_ACCESS_KEY_ID, safe=''),
    AWS_SECRET_ACCESS_KEY=quote(AWS_SECRET_ACCESS_KEY, safe='')
)

【讨论】：