使用来自 Celery 的 SQS 消息答案

【问题标题】：Consume SQS messages from Celery使用来自 Celery 的 SQS 消息
【发布时间】：2019-04-01 13:29:14
【问题描述】：

我的 Python 应用程序中有一个 Celery 实现。我使用的代理是 SQS。通过 Boto3 的 send_message() api 发送到 SQS 的消息来自不同的应用程序。现在我的困惑是如何触发 Celery 从 SQS 中挑选消息进行处理。将有一些任务将在 Celery 中运行，它应该正确处理来自 SQS 的消息。我的要求类似于Celery Consumer SQS Messages。

据我了解，Celery 会轮询 SQS，直到消息到达那里。有人可以帮我解决这个问题吗？

【问题讨论】：

想知道您是否可以使用 boto3 运行 Celery 任务。我正在尝试相同的方法，甚至在 boto3 github repo [1] 中开始和问题/问题。与此同时，我正在使用 Celery 本身从远程客户端启动任务：python app = Celery(broker='sqs://access_key:secret_key@sqs_server:9324') app.send_task('tasks.my_task', args=[1], queue='my_queue') [1]：github.com/boto/boto3/issues/2037

标签： python amazon-sqs django-celery

【解决方案1】：

我每 20 秒调用一次这个任务：

@app.task(name='listen_to_sqs_telemetry')
def listen_to_sqs_telemetry():
    logger.info('start listen_to_telemetry')
    sqs = get_sqs_client()
    queue_url = 'https://sqs.us-east-2.amazonaws.com/xxx'
    logger.info('Using ' + queue_url)

    keep_going = True
    num = 0
    while keep_going:
        keep_going = False
        try:
            response = sqs.receive_message(
                QueueUrl=queue_url,
                AttributeNames=[
                    'SentTimestamp',
                ],
                MaxNumberOfMessages=5,
                MessageAttributeNames=[
                    'All'
                ],
                WaitTimeSeconds=20
            )
            # logger.info(response)
            if 'Messages' in response:
                keep_going = True
                for rec in response['Messages']:
                    # Process message
                    sqs.delete_message(
                        QueueUrl=queue_url,
                        ReceiptHandle=rec['ReceiptHandle']
                    )
                    num = num + 1
            else:
                pass
                # logger.info(response)
        except Exception as e:
            logger.error(str(e))
    logger.info('done with listen_to_sqs_telemetry')
    return "Processed {} message(s)".format(num)

【讨论】：

【解决方案2】：

如果我理解你，请尝试将工作程序作为守护进程运行。使用像 supervisord 这样的工具来做。

【讨论】：