在芹菜SQS消费者上填充不正确

Question

我正在尝试为SQS队列设置芹菜消费者。

我正在使用python3使用芹菜4.1.0，然后我在库boto3(1.5.28(中发送了一条消息：

response = queue.send_message(MessageBody='Hello World')

到目前为止一切都可以。

然后我尝试以这种方式设置芹菜消费者：

@app.task(base=celery.Task, name='test', bind=True)
def test(self, message):
    print(message)
    return True

但是我得到了这个追溯：

  ...
  File ".../lib/python3.6/site-packages/kombu/transport/SQS.py", line 350, in _on_messages_ready
    msg_parsed = self._message_to_python(msg, qname, queue)
  File ".../lib/python3.6/site-packages/kombu/transport/SQS.py", line 215, in _message_to_python
    body = base64.b64decode(message['Body'].encode())
  File ".../lib/python3.6/base64.py", line 87, in b64decode
    return binascii.a2b_base64(s)
binascii.Error: Incorrect padding

我做了什么错了吗？我是否需要一些特定的配置来与SQS队列一起使用？

预先感谢！

Answer 1

您的消息需要为基本64编码JSON：

message = {"test":"test"}
message_string = json.dumps(message)
byte_message = base64.b64encode(message_string.encode('utf-8'))
base64_json_string = byte_message.decode()
response = queue.send_message(MessageBody=base64_json_string)