Django-Celery-SQS-S3-接收消息

Question

我有一个Django应用程序，我使用的是Celery、SQS和S3。当我使用Django、Celery和SQS运行以下函数时，该函数会正常工作，并且它会每分钟打印一次"hello"。

from celery.task import periodic_task
from celery.schedules import crontab
@periodic_task(run_every=crontab(hour='*', minute='*', day_of_week="*"))
def print_hello():
print('hello world')

但该应用程序也链接到S3 Bucket。每当将新文件保存到S3时，都会向SQS队列发送通知。将通知消息发送到SQS队列时会出现此问题。当通知到达队列时，工作进程将失败。它停止定期任务print_hello((，给出以下错误消息：

〔2019-11-07 22:10:57173：关键/主进程〕不可恢复的错误：错误("填充错误"(…parserinvoker/lib64/python3.7/base64.py"，b64decode中的第87行返回binascii.a2b_base64(s(binascii。错误：填充不正确

，然后退出。我一直在查看文档，整个星期都在尝试进行故障排除，但没有找到解决方案。我正在包括我的设置.py，以防出现配置问题

设置.py

BROKER_URL = "sqs://"
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_DEFAULT_QUEUE = env('CELERY_DEFAULT_QUEUE')
CELERY_RESULT_BACKEND = None 
BROKER_TRANSPORT_OPTIONS = {
'region': 'us-east-1',
'polling_interval':20,
'visibility_timeout': 3600,
'task_default_queue': env('CELERY_DEFAULT_QUEUE'),
}

Answer 1

芹菜在队列中期望的json有效载荷的格式与SQS从s3接收的格式不同；为了正确处理这些问题，您可能需要一个单独的定期任务来定期检查这些问题，并耗尽s3通知队列，而不是将s3通知发送到celebroker队列。s3消息体的外观如amazon文档中所述。以下是从S3发送到SQS:的2.1记录样本

"Records":[  
{  
"eventVersion":"2.1",
"eventSource":"aws:s3",
"awsRegion":"us-west-2",
"eventTime":The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, when Amazon S3 finished processing the request,
"eventName":"event-type",
"userIdentity":{  
"principalId":"Amazon-customer-ID-of-the-user-who-caused-the-event"
},
"requestParameters":{  
"sourceIPAddress":"ip-address-where-request-came-from"
},
"responseElements":{  
"x-amz-request-id":"Amazon S3 generated request ID",
"x-amz-id-2":"Amazon S3 host that processed the request"
},
"s3":{  
"s3SchemaVersion":"1.0",
"configurationId":"ID found in the bucket notification configuration",
"bucket":{  
"name":"bucket-name",
"ownerIdentity":{  
"principalId":"Amazon-customer-ID-of-the-bucket-owner"
},
"arn":"bucket-ARN"
},
"object":{  
"key":"object-key",
"size":object-size,
"eTag":"object eTag",
"versionId":"object version if bucket is versioning-enabled, otherwise null",
"sequencer": "a string representation of a hexadecimal value used to determine event sequence, 
only used with PUTs and DELETEs"
}
},
"glacierEventData": {
"restoreEventData": {
"lifecycleRestorationExpiryTime": "The time, in ISO-8601 format, for example, 1970-01-01T00:00:00.000Z, of Restore Expiry",
"lifecycleRestoreStorageClass": "Source storage class for restore"
}
}
}
]
}

芹菜消息格式如下所示。