以下是使用AWS和SQS实现异步任务的最佳实践。这个示例展示一个基本的生产者/消费者模型。

首先，我们需要创建一个SQS队列，并为应用程序分配必要的IAM权限。

import boto3

sqs_client = boto3.client('sqs', region_name='us-east-1')

queue_name = 'my_queue'
queue_url = sqs_client.create_queue(QueueName=queue_name)['QueueUrl']

# grant permissions to access the queue to our application
policy = {
    'Version': '2012-10-17',
    'Id': 'example-policy',
    'Statement': [
        {
            'Sid': 'example-statement',
            'Effect': 'Allow',
            'Principal': {
                'AWS': 'arn:aws:iam::123456789012:root'
            },
            'Action': 'sqs:*',
            'Resource': queue_url
        }
    ]
}

sqs_client.set_queue_attributes(
    QueueUrl=queue_url,
    Attributes={
        'Policy': json.dumps(policy)
    }
)

接下来，我们将编写一个生产者函数，并将消息发送到队列中。

def send_message_to_sqs_queue(queue_url, message_body):
    sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=message_body
    )

然后，我们编写一个消费者函数，用于执行从队列中接收到的任务。

def process_message_from_sqs_queue(queue_url):
    response = sqs_client.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=1,
        WaitTimeSeconds=20
    )
    
    if 'Messages' not in response:
        return
    
    message = response['Messages'][0]
    # process the message here
    
    sqs_client.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=message['ReceiptHandle']
    )

最后，我们可以将这些函数连接起来，并使应用程序在后台执行。

from threading import Thread

queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my_queue'

def producer(queue_url):
    while True:
        # get task to execute
        task = get_task()
        
        # send the task to the SQS queue
        send_message_to_sqs_queue(queue_url, task)
            
def consumer(queue_url):
    while True:
        process_message_from_sqs_queue(queue_url)

if __name__ == '__main__':