App Engine Flex实例崩溃：线程问题还是/tmp文件存储导致？

现有代码的核心问题 #

• gunicorn多worker导致全局队列混乱：配置了workers = multiprocessing.cpu_count() * 2 + 1，每个worker是独立进程，全局jobs = Queue()会被每个worker复制一份，任务逻辑完全混乱
• 任务提前执行，队列沦为摆设：some_route里直接调用heavy_func再把返回值丢进队列，等于任务已经在请求线程里执行完了，后面开的线程根本没做实际工作，白开线程还占资源
• 线程安全隐患：用while not q.empty()判断任务是否结束不是线程安全的，可能出现任务还没处理完线程就退出的情况
• 请求阻塞触发超时崩溃：在请求路由里调用jobs.join()会阻塞请求直到所有任务完成，一旦任务量较大就会超过请求超时阈值，直接导致实例崩溃

改用Google Cloud Tasks的实现方案 #

Cloud Tasks是分布式队列服务，把任务异步提交到队列后，由独立的处理路由消费执行，完全规避本地线程/队列的适配问题，符合App Engine的环境限制。

步骤1：创建Cloud Tasks队列 #

先在Google Cloud控制台创建队列（比如命名为task-queue），或用gcloud命令快速创建：

gcloud tasks queues create task-queue

步骤2：调整app.yaml配置 #

补充Cloud Tasks相关环境变量，确保服务能访问队列：

runtime: python
env: flex
instance_class: F2
runtime_config:
  python_version: 3.7
env_variables:
  CLOUD_SQL_USERNAME: "my username"
  CLOUD_SQL_PASSWORD: "pass"
  CLOUD_SQL_DATABASE_NAME: "db"
  CLOUD_SQL_CONNECTION_NAME: "conn"
  CLOUD_TASKS_PROJECT: "你的GCP项目ID"
  CLOUD_TASKS_LOCATION: "队列所在区域，比如us-central1"
  CLOUD_TASKS_QUEUE: "task-queue"
entrypoint: gunicorn -c gunicorn.conf.py -b :8080 main:app --log-level=DEBUG --timeout=600
automatic_scaling:
  min_num_instances: 1
  max_num_instances: 8
  cpu_utilization:
    target_utilization: 0.6
beta_settings:
  cloud_sql_instances: "你的Cloud SQL连接名"

步骤3：gunicorn配置保持不变 #

import multiprocessing

workers = multiprocessing.cpu_count() * 2 + 1

步骤4：重写main.py核心逻辑 #

拆分任务提交与任务处理流程，用Cloud Tasks替代本地队列：

from flask import Flask, request, jsonify
from google.cloud import tasks_v2
import os
import json
import requests

app = Flask(__name__)

# 初始化Cloud Tasks客户端
client = tasks_v2.CloudTasksClient()
PROJECT_ID = os.environ.get("CLOUD_TASKS_PROJECT")
LOCATION = os.environ.get("CLOUD_TASKS_LOCATION")
QUEUE = os.environ.get("CLOUD_TASKS_QUEUE")
QUEUE_PATH = client.queue_path(PROJECT_ID, LOCATION, QUEUE)

def heavy_func(object_data, text_file, mp3_file):
    """实际任务逻辑：下载文件、处理、保存到/tmp后删除"""
    try:
        # 从指定URL下载文件
        response = requests.get(object_data['url'])
        tmp_file_path = f"/tmp/{mp3_file}"
        
        # 写入临时文件
        with open(tmp_file_path, 'wb') as f:
            f.write(response.content)
        
        # 这里添加你的文件处理逻辑（比如重命名、格式转换等）
        
        # 处理完成后立即删除临时文件
        os.remove(tmp_file_path)
        return "任务处理完成"
    except Exception as e:
        print(f"任务执行失败：{str(e)}")
        raise  # 抛出异常让Cloud Tasks自动重试

@app.route('/submit-tasks', methods=['POST'])
def submit_tasks():
    # 从请求获取任务参数
    request_data = request.get_json()
    string_list = request_data.get('string_list', [])
    object_data = request_data.get('object')
    text_file = request_data.get('text_file')
    base_mp3_file = request_data.get('mp3_file')
    
    # 批量提交任务到Cloud Tasks
    for item in string_list:
        # 构造任务参数（序列化可传输的数据）
        task_payload = json.dumps({
            'object': object_data,
            'text_file': text_file,
            'mp3_file': f"{item}_{base_mp3_file}"  # 给每个任务的文件名加前缀，避免冲突
        }).encode()
        
        # 定义任务请求
        task = {
            'http_request': {
                'http_method': tasks_v2.HttpMethod.POST,
                'url': f"{request.host_url}process-task",
                'body': task_payload,
                'headers': {'Content-Type': 'application/json'}
            }
        }
        
        # 发送任务到队列
        client.create_task(request={"parent": QUEUE_PATH, "task": task})
    
    return jsonify({"status": "success", "message": "所有任务已提交到Cloud Tasks队列"})

@app.route('/process-task', methods=['POST'])
def process_task():
    # 处理Cloud Tasks推送的任务
    payload = request.get_json()
    heavy_func(payload['object'], payload['text_file'], payload['mp3_file'])
    return jsonify({"status": "success"}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

关键注意事项 #

• /tmp目录使用规范：App Engine Flex的/tmp是实例本地临时存储，每个实例有独立的/tmp，任务处理完成后必须删除文件，避免磁盘耗尽
• 任务幂等性设计：Cloud Tasks会自动重试失败任务，heavy_func要保证重复执行不会产生错误数据（比如下载前先检查文件是否存在）
• 超时控制：默认Cloud Tasks任务超时为10分钟，可在创建任务时通过schedule_time或timeout参数调整
• 权限配置：给App Engine服务账号添加Cloud Tasks Enqueuer和Cloud Tasks Consumer角色，确保能操作队列

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