3

我在 LocalStack 服务器上有一个 SQS 队列，我正在尝试使用 Celery 消费者从中获取消息。

它表明消费者已正确附加到队列，例如队列，但是当我尝试使用命令sqs-test-queue发送消息时它没有收到任何消息。aws

我的celeryconfig.py样子是这样的：

from kombu import (
    Exchange,
    Queue
)


broker_transport_options = {'region': REGION}
broker_transport = 'sqs'

accept_content = ['application/json']
result_serializer = 'json'
content_encoding = 'utf-8'
task_serializer = 'json'

worker_enable_remote_control = False
worker_send_task_events = False
result_backend = None

task_queues = (
    Queue('sqs-test-queue', exchange=Exchange(''), routing_key='sqs-test-queue'),
)

我的tasks.py模块看起来像这样：

from celery import Celery
from kombu.utils.url import quote


AWS_ACCESS_KEY = quote("AWS_ACCESS_KEY")
AWS_SECRET_KEY = quote("AWS_SECRET_KEY")
LOCALSTACK = "<IP>:<PORT>"

broker_url = "sqs://{access}:{secret}@{host}".format(access=AWS_ACCESS_KEY,
                                                     secret=AWS_SECRET_KEY,
                                                     host=LOCALSTACK)

app = Celery('tasks', broker=broker_url, backend=None)
app.config_from_object('celeryconfig')


@app.task(bind=True, name='tasks.consume', acks_late=True, ignore_result=True)
def consume(self, msg):
    # DO SOMETHING WITH THE RECEIVED MESSAGE
    return True

试图用它来执行它，celery -A tasks worker -l INFO -Q sqs-test-queue一切都很好：

...

[tasks]
  . tasks.consume

[... INFO/MainProcess] Connected to sqs://AWS_ACCESS_KEY:**@<IP>:<PORT>// 
[... INFO/MainProcess] celery@local ready

但是当我尝试用 发送消息时aws sqs send-message --endpoint-url=http://<IP>:<PORT> --queue-url=http://localhost:<PORT>/queue/sqs-test-queue --message-body="Test message"，什么也没有发生。

我究竟做错了什么？我是否错过了配置中的某些内容？

PS：如果我尝试运行命令aws sqs receive-message --endpoint-url=http://<IP>:<PORT> --queue-url=http://localhost:<PORT>/queue/sqs-test-queue，我可以得到消息。

笔记：

我正在使用Python 3.7.0，我的pip freeze样子是这样的：

boto3==1.10.16
botocore==1.13.16
celery==4.3.0
kombu==4.6.6
pycurl==7.43.0.3
...
4

2 回答 2

3

我正在经历和你一样的事情。为了解决这个问题，我做了几件事：

  • 我在 localstack 中设置HOSTNAME_EXTERNAL和env 变量HOSTNAME
  • 设置broker_url为sqs://{access}:{secret}@{host}:{port}（如你所愿）
  • 确保 celery workerbroker_transport_options不包含配置项：wait_time_seconds因为这会导致 2020 年 2 月 7 日的 localstack 出现错误（在此处查看问题）。

一旦我做了这两件事，它就开始工作了，希望它有所帮助。

于 2020-02-07T15:00:29.477 回答
1

Celery 不能向/从任何消息队列系统发布或使用任意消息。为此使用kombu - 这也是Celery在幕后使用的。

于 2019-11-13T10:31:30.367 回答