咨询Python实现AWS Lambda启动异步进程无需等待直接返回的方法

Lambda运行环境会在handler函数返回后立即冻结整个实例，未执行完成的本地代码（包括普通Python异步协程、后台线程）都会直接终止，因此无法通过Python原生异步语法实现需求。你需要拆分为两个Lambda函数配合DynamoDB实现，具体步骤如下：

前置配置 #

你需要先在AWS控制台完成以下基础配置：

• 创建两个Lambda函数，分别命名为job_entry_lambda（入口请求处理）、long_task_lambda（长任务执行）
• 给job_entry_lambda的执行角色添加lambda:InvokeFunction权限，允许它调用long_task_lambda
• 创建DynamoDB表，主键设置为job_id（字符串类型），给两个Lambda的执行角色添加对应表的读写权限
• 给两个Lambda配置环境变量DYNAMODB_TABLE_NAME，值为你创建的DynamoDB表名

入口Lambda代码（job_entry_lambda） #

负责接收用户请求、生成jobID、预写任务状态、异步触发长任务后直接返回jobID：

import uuid
import boto3
import os
import json

# 初始化AWS服务客户端
lambda_client = boto3.client('lambda')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE_NAME'])

def lambda_handler(event, context):
    # 生成唯一任务ID
    job_id = uuid.uuid4().hex
    # 读取用户提交的参数
    stuff = event['stuff']
    
    # 预写任务初始状态到DynamoDB
    table.put_item(
        Item={
            'job_id': job_id,
            'status': 'pending',
            'result': None,
            'stuff': stuff
        }
    )
    
    # 异步调用长任务Lambda，InvocationType设为Event即代表异步调用，调用后立即返回无需等待执行完成
    lambda_client.invoke(
        FunctionName='long_task_lambda', # 替换为你的长任务Lambda实际名称
        InvocationType='Event',
        Payload=json.dumps({'job_id': job_id, 'stuff': stuff})
    )
    
    # 直接返回jobID给用户
    return {
        'statusCode': 200,
        'body': json.dumps({'job_id': job_id})
    }

长任务Lambda代码（long_task_lambda） #

负责执行耗时逻辑，执行完成后更新任务状态和结果到DynamoDB：

import boto3
import os
import time

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE_NAME'])

def long_function(job_id, stuff):
    # 此处替换为你的实际慢执行逻辑
    time.sleep(30) # 模拟30秒耗时操作
    result = f"处理完成，输入参数为：{stuff}"
    return result

def lambda_handler(event, context):
    job_id = event['job_id']
    stuff = event['stuff']
    
    try:
        # 执行长任务逻辑
        result = long_function(job_id, stuff)
        # 更新任务状态为成功，写入运行结果
        table.update_item(
            Key={'job_id': job_id},
            UpdateExpression="set #s = :s, result = :r",
            ExpressionAttributeNames={'#s': 'status'},
            ExpressionAttributeValues={
                ':s': 'success',
                ':r': result
            }
        )
    except Exception as e:
        # 任务执行失败时更新状态为失败，写入错误信息
        table.update_item(
            Key={'job_id': job_id},
            UpdateExpression="set #s = :s, error_msg = :e",
            ExpressionAttributeNames={'#s': 'status'},
            ExpressionAttributeValues={
                ':s': 'failed',
                ':e': str(e)
            }
        )
    return

任务查询逻辑补充 #

你可以在入口Lambda中增加GET请求处理逻辑，支持用户通过jobID查询任务状态和结果：

# 将以下逻辑加入job_entry_lambda的lambda_handler函数开头
if event.get('httpMethod') == 'GET':
    job_id = event['queryStringParameters']['job_id']
    resp = table.get_item(Key={'job_id': job_id})
    if 'Item' not in resp:
        return {'statusCode': 404, 'body': '任务不存在'}
    return {'statusCode': 200, 'body': json.dumps(resp['Item'])}

内容的提问来源于stack exchange，提问作者Edward Peters