基于AWS Lambda实现用户级并发控制的异步任务队列架构咨询

核心问题拆解 #

你需要实现多任务并行执行，但同一用户的任务必须串行——Lambda本身没有按用户维度管控并发的原生机制，但可以通过AWS服务组合来复刻现有Celery多队列的逻辑。

具体架构调整方案 #

1. 保留SQS作为任务队列，用消息属性替代用户专属队列 #

• 无需为每个用户创建独立队列，只需在SQS消息中添加user_id作为消息属性。
• 结合SQS消息可见性超时+DynamoDB状态锁实现单用户串行：
  • Lambda处理任务前，先检查该用户是否有正在执行的任务；
  • 若有，将当前消息重新放回队列并设置延迟（比如30秒），避免重复触发；
  • 若无，标记用户为"执行中"，处理完成后再释放状态。

2. 用DynamoDB做任务锁 #

• 创建DynamoDB表，主键设为user_id，字段包含is_processing（布尔值）、task_id；
• Lambda处理任务前，用条件表达式更新表：仅当is_processing = false时，才将其设为true；
• 更新成功则执行任务，失败则说明该用户已有任务在运行，直接将消息延迟重入队。

3. 用EventBridge替代Celery Beat #

• 配置Amazon EventBridge规则，按需求的时间间隔触发Lambda，由Lambda调用Django任务逻辑或直接向SQS发送定时任务消息。

4. Lambda配置优化 #

• 设置合理的并发上限，避免突发任务导致成本超支；
• 开启Lambda异步调用模式，让Django应用快速返回，任务由Lambda后台处理。

关键代码示例（伪代码） #

import boto3
import json
from my_django_app.tasks import process_user_task

dynamodb = boto3.resource('dynamodb')
task_lock_table = dynamodb.Table('user_task_locks')
sqs = boto3.client('sqs')
SQS_QUEUE_URL = "your-sqs-queue-url"

def lambda_handler(event, context):
    for record in event['Records']:
        message_body = json.loads(record['body'])
        user_id = message_body['user_id']
        task_params = message_body['params']
        
        # 尝试获取用户任务锁
        try:
            task_lock_table.update_item(
                Key={'user_id': user_id},
                UpdateExpression='SET is_processing = :processing',
                ConditionExpression='attribute_not_exists(is_processing) OR is_processing = :idle',
                ExpressionAttributeValues={':processing': True, ':idle': False},
                ReturnValues='ALL_NEW'
            )
            # 获取锁成功，执行任务
            process_user_task(task_params)
            # 任务完成，释放锁
            task_lock_table.update_item(
                Key={'user_id': user_id},
                UpdateExpression='SET is_processing = :idle',
                ExpressionAttributeValues={':idle': False}
            )
        except task_lock_table.exceptions.ConditionalCheckFailedException:
            # 获取锁失败，延迟重发消息
            sqs.send_message(
                QueueUrl=SQS_QUEUE_URL,
                MessageBody=json.dumps(message_body),
                DelaySeconds=30
            )
    return {'statusCode': 200}

成本优化适配 #

• Lambda按执行时长和调用次数计费，空闲时段无成本，完美匹配你应用的批量任务涌入特性；
• SQS标准队列按请求次数计费，相比EC2固定成本，波动型任务场景下成本更低；
• DynamoDB采用按需模式，仅读写操作产生费用，适配任务量波动的场景。

内容的提问来源于stack exchange，提问作者David Carli-Arnold