gpt4 book ai didi

django - 如何在 SQS 中解码 celery 消息

转载 作者:行者123 更新时间:2023-12-05 04:05:07 25 4
gpt4 key购买 nike

sqs 中的某些 celery 任务永远挂起,我想在删除之前阅读这些消息(任务)。在转到 sqs 控制台时,我能够看到我尝试用它解码的编码消息

value = base64.b64decode(value.encode('utf-8')).decode('utf-8')

这给了我带键的字典转储

['body', 'headers', 'content-type', 'properties', 'content-encoding']

在这个字典中,正文看起来像编码的我试着用同样的方式解码它

value = base64.b64decode(value.encode('utf-8')).decode('utf-8')

但它给出了错误的说法UnicodeDecodeError:“utf8”编解码器无法解码位置 1 中的字节 0x87:起始字节无效

我错过了什么吗?如何解码此消息?有什么办法可以解码吗?

最佳答案

似乎“Celery”使用“pickle.dump”将任务的有效负载转换为字节,然后编码为base64。进行反向操作,我们再次获得有效载荷。

import base64
import boto3
import pickle

queue_name = 'your-queue-name'
sqsr = boto3.resource('sqs')
queue = sqsr.get_queue_by_name(QueueName=queue_name)

for message in queue.receive_messages(MaxNumberOfMessages=10):
print(f'{message.message_id} >>> {message.receipt_handle}'
f' >>> {message.body} >>> {message.message_attributes}')
body_dict = json.loads(base64.b64decode(message.body))
celery_payload = pickle.loads(base64.b64decode(body_dict.get('body')))
print(celery_payload)

关于django - 如何在 SQS 中解码 celery 消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51515692/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com