如何使用 celery worker 从 SQS 轮询消息，消息为 JSON 格式，并且 worker 无法解码该格式

Posted 2023-02-24

【中文标题】如何使用 celery worker 从 SQS 轮询消息，消息为 JSON 格式，并且 worker 无法解码该格式

【英文标题】：how to poll message from SQS using celery worker, the message is in JSON format and worker not able to decode the format

【发布时间】：2019-12-14 16:05:51

【问题描述】：

如何使用 celery worker 从 SQS 轮询消息，消息是 JSON 格式，worker 无法解码格式

注意：这些消息不是使用celery beat发送到SQS的，这个队列是从SNS订阅的

我的 celery worker 命令是： celery worker -A status_handling -l info -Q es_status_test

Msg in Queue:


  "Type" : "Notification",
  "MessageId" : "f7e40fd9-8f92-59c5-afd9-5a1847aaae57",
  "TopicArn" : "***",
  "Message" : "\"SESResponseStatusCode\": 200, \"Status\": \"Delivered\", \"Message\": \"Email sent successfully.\", \"MessageId\": \"a59e85a2-8b7a-4b49-9354-0a7a4170b0c0\", \"Uuid\": null",
  "Timestamp" : "2019-08-05T06:00:24.943Z",
  "SignatureVersion" : "1",
  "Signature" : "pass",
  "SigningCertURL" : "pass",
  "UnsubscribeURL" : "pass"

错误来了：

[2019-08-04 23:00:25,116: CRITICAL/MainProcess] Unrecoverable error: JSONDecodeError('Expecting value: line 1 column 1 (char 0)')
Traceback (most recent call last):
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/worker/worker.py", line 205, in start
    self.blueprint.start(self)
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/bootsteps.py", line 369, in start
    return self.obj.start()
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 596, in start
    c.loop(*c.loop_args())
  File "/home/vagrant/env/lib/python3.7/site-packages/celery/worker/loops.py", line 91, in asynloop
    next(loop)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/asynchronous/hub.py", line 362, in create_loop
    cb(*cbargs)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/asynchronous/http/curl.py", line 111, in on_readable
    return self._on_event(fd, _pycurl.CSELECT_IN)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/asynchronous/http/curl.py", line 124, in _on_event
    self._process_pending_requests()
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/asynchronous/http/curl.py", line 130, in _process_pending_requests
    self._process(curl)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/asynchronous/http/curl.py", line 178, in _process
    buffer=buffer, effective_url=effective_url, error=error,
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 177, in __call__
    svpending(*ca, **ck)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 170, in __call__
    return self.throw()
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 167, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/funtools.py", line 100, in _transback
    return callback(ret)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 170, in __call__
    return self.throw()
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 167, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/funtools.py", line 100, in _transback
    return callback(ret)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 170, in __call__
    return self.throw()
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/promises.py", line 167, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/funtools.py", line 98, in _transback
    callback.throw()
  File "/home/vagrant/env/lib/python3.7/site-packages/vine/funtools.py", line 96, in _transback
    ret = filter_(*args + (ret,), **kwargs)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/transport/SQS.py", line 370, in _on_messages_ready
    msg_parsed = self._message_to_python(msg, qname, queue)
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/transport/SQS.py", line 236, in _message_to_python
    payload = loads(bytes_to_str(body))
  File "/home/vagrant/env/lib/python3.7/site-packages/kombu/utils/json.py", line 94, in loads
    return stdjson.loads(s)
  File "/usr/local/lib/python3.7/json/__init__.py", line 348, in loads
    return _default_decoder.decode(s)
  File "/usr/local/lib/python3.7/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/lib/python3.7/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

【参考方案1】：

Celery 中的 SQS 支持仅用作 Celery 特定消息的传输机制。您不能使用 Celery 来消费任意 SQS 消息。

相反，我建议编写一个自定义 Django 管理命令，您可以在其中使用 boto3 库轮询 SQS 队列。