仅用AWS Lambda（Python）能否实现游戏临时排行榜？

可行性结论 #

可以实现，无需依赖任何外部服务，只需结合Lambda的本地存储特性与客户端数据传递即可满足需求。

核心思路 #

Lambda多实例的独立内存导致全局变量不可靠，但我们可以通过以下方式规避：

• 利用Lambda实例的/tmp临时存储，在实例复用期间保留数据
• 让客户端每次提交得分时附带之前获取的全局得分数据，实现跨实例的数据合并
• 接受中间排名的不准确性，确保所有提交完成后，最终数据能被完整合并得到正确排名

具体实现代码 #

基础版本（依赖客户端数据传递） #

import json

def lambda_handler(event, context):
    # 解析请求参数
    request_body = json.loads(event['body'])
    player_id = request_body['player_id']
    new_score = request_body['score']
    existing_scores = request_body.get('existing_scores', {})
    
    # 合并数据：保留每个玩家的最新得分（可根据需求改为累计得分）
    merged_scores = existing_scores.copy()
    merged_scores[player_id] = new_score
    
    # 生成当前排名
    sorted_ranking = sorted(merged_scores.items(), key=lambda x: x[1], reverse=True)
    ranking = [
        {'player_id': pid, 'score': s, 'rank': idx + 1}
        for idx, (pid, s) in enumerate(sorted_ranking)
    ]
    
    # 返回结果，附带合并后的完整得分数据供客户端下次提交使用
    return {
        'statusCode': 200,
        'body': json.dumps({
            'ranking': ranking,
            'existing_scores': merged_scores
        })
    }

增强版本（结合/tmp本地存储） #

增加本地存储提升数据可靠性，避免客户端丢失数据导致的得分遗漏：

import json
import os
import fcntl

# 本地存储文件路径
TMP_SCORE_FILE = '/tmp/game_scores.json'

def lambda_handler(event, context):
    # 解析请求参数
    request_body = json.loads(event['body'])
    player_id = request_body['player_id']
    new_score = request_body['score']
    client_provided_scores = request_body.get('existing_scores', {})
    
    # 读取本地存储的得分数据（加锁避免同一实例内并发冲突）
    local_scores = {}
    if os.path.exists(TMP_SCORE_FILE):
        with open(TMP_SCORE_FILE, 'r') as f:
            fcntl.flock(f, fcntl.LOCK_SH)
            local_scores = json.load(f)
            fcntl.flock(f, fcntl.LOCK_UN)
    
    # 合并本地数据与客户端数据：优先保留客户端的最新数据
    merged_scores = {**local_scores, **client_provided_scores}
    merged_scores[player_id] = new_score
    
    # 将合并后的数据写回本地存储（加锁确保原子性）
    with open(TMP_SCORE_FILE, 'w') as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        json.dump(merged_scores, f)
        fcntl.flock(f, fcntl.LOCK_UN)
    
    # 生成排名
    sorted_ranking = sorted(merged_scores.items(), key=lambda x: x[1], reverse=True)
    ranking = [
        {'player_id': pid, 'score': s, 'rank': idx + 1}
        for idx, (pid, s) in enumerate(sorted_ranking)
    ]
    
    # 返回结果
    return {
        'statusCode': 200,
        'body': json.dumps({
            'ranking': ranking,
            'existing_scores': merged_scores
        })
    }

方案说明 #

• 最终正确性：当所有玩家停止提交后，最后一次请求会合并所有历史数据，生成正确排名
• 实时性妥协：并行提交时可能出现部分数据延迟同步，导致中间排名不准确，但符合需求
• 数据生命周期：Lambda实例超时销毁后，/tmp数据会丢失，无需长期存储，符合要求
• 客户端要求：需要客户端每次提交时携带上一次返回的existing_scores，确保跨实例的数据同步

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