具有 Celery 配置的 SQS

Question

我尝试为 Celery 设置 Amazon SQS，我有以下配置：

BROKER_BACKEND = "SQS"
BROKER_TRANSPORT_OPTIONS = {
    'region': 'us-east-1',
}
AWS_ACCESS_KEY_ID = # access id
AWS_SECRET_ACCESS_KEY = # secret access key
os.environ.setdefault("AWS_ACCESS_KEY_ID", AWS_ACCESS_KEY_ID)
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY)

BROKER_URL = 'sqs://'

CELERY_IMPORTS = ("tasks", )
CELERY_TASK_RESULT_EXPIRES = 300

CELERY_DEFAULT_QUEUE = #queue name
CELERY_DEFAULT_EXCHANGE = CELERY_DEFAULT_QUEUE
CELERY_DEFAULT_EXCHANGE_TYPE = CELERY_DEFAULT_QUEUE
CELERY_DEFAULT_ROUTING_KEY = CELERY_DEFAULT_QUEUE
CELERY_QUEUES = {
    CELERY_DEFAULT_QUEUE: {
        'exchange': CELERY_DEFAULT_QUEUE,
        'binding_key': CELERY_DEFAULT_QUEUE,
    }
}

在 AWS 账户上的 SQS 配置中，我有一个名称为

CELERY_DEFAULT_QUEUE

的队列。当我在本地运行此命令时，一切正常...但由于某种原因，它在 SQS 上创建了另一个名称格式为

<user_id>-celery-pidbox

的队列。像这样的：

MyUser-MacBook-Pro-local-celery-pidbox

。

这正常吗？当我使用指定名称创建了一个队列时，为什么它会创建另一个队列？否则，它可以工作，不确定是否需要其他队列或者我错过了什么？感谢任何帮助，我在文档中找不到这个。

编辑

事实证明这是正常的。由于某种原因，django-celery 会这样做，它为您访问要访问的队列的每个框创建一个队列。他们将在未来的版本中修复此问题。如果有人知道如何暂时解决此问题，请告诉我，谢谢！

Answer 1

这实际上是一个很好的行为，因此您可以监控哪些实例（IP 或本地名称）正在访问您的 SQS 帐户。这只是一个请求，因此不会花费您任何费用。

Answer 2

您需要设置这些：

 CELERY_ENABLE_REMOTE_CONTROL = False 
 CELERY_SEND_EVENTS = False

禁用它。

Answer 3

如果你想将 Celery 与 SQS 连接，那么你应该使用以下代码创建一个 celery 应用程序

from celery import Celery
def make_celery(app):
    celery = Celery(
        app.import_name,
        broker="sqs://",
        broker_transport_options={
            "queue_name_prefix": "{SERVICE_ENV}-{SERVICE_NAME}-"
        },
    )
    task_base = celery.Task

    class ContextTask(task_base):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return task_base.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

使用此代码，您将能够将 Celery 与 SQS 连接。