在Flask中运行Python并实现持续后台任务的GCP部署问题

Got it, let's break this down. App Engine's Python 3.6 standard environment is built around request-driven execution, so a naive "run forever" loop will get shut down when there's no incoming traffic. Here are three practical approaches to get your persistent background task working while keeping a route to view its output:

1. Lightweight Background Thread (For Non-Critical Tasks) #

If your task is low-stakes and can tolerate occasional restarts, you can spin up a background thread in your Flask app. Just keep in mind that App Engine might scale down instances if there's no traffic, so you'll need to tweak your instance settings to keep it running.

Here's a working example:

from flask import Flask
import threading
import time
from datetime import datetime

app = Flask(__name__)

# Store task output in memory (use Cloud Datastore/Memorystore for persistence)
task_logs = []

def persistent_task():
    """Your long-running background task"""
    while True:
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"Task running at {timestamp}"
        task_logs.append(log_entry)
        
        # Trim old logs to avoid memory bloat
        if len(task_logs) > 100:
            task_logs.pop(0)
        
        time.sleep(5)  # Adjust interval as needed

# Start the thread when the app initializes
threading.Thread(target=persistent_task, daemon=True).start()

@app.route('/')
def hello():
    return 'Hello, world!'

@app.route('/task-logs')
def view_logs():
    """Route to display task output"""
    return '<br>'.join(task_logs)

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

Key Notes: #

• Add this to your app.yaml to prevent instance shutdown:

  runtime: python36
  automatic_scaling:
    min_instances: 1  # Keep at least one instance running
  

• The in-memory task_logs will reset if the instance restarts. For persistent logs, use Cloud Datastore or Cloud Memorystore (Redis) instead.

2. Cloud Tasks + Persistent Storage (For Reliable Tasks) #

If your task needs to run reliably (no missed executions even if instances restart), use Cloud Tasks to trigger your task on a schedule, and store output in a persistent service like Cloud Datastore.

Step-by-Step Implementation: #

1. Create a Cloud Tasks queue in the GCP Console.
2. Update your Flask app to handle task triggers and store logs:

from flask import Flask, request
from google.cloud import datastore
from datetime import datetime

app = Flask(__name__)
ds_client = datastore.Client()

@app.route('/')
def hello():
    return 'Hello, world!'

@app.route('/execute-task', methods=['POST'])
def run_task():
    """Endpoint triggered by Cloud Tasks"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"Task executed at {timestamp}"
    
    # Save log to Cloud Datastore
    log_entity = datastore.Entity(key=ds_client.key('TaskLog'))
    log_entity.update({
        'message': log_entry,
        'timestamp': datetime.now()
    })
    ds_client.put(log_entity)
    
    return "Task completed", 200

@app.route('/task-logs')
def view_logs():
    """Fetch and display stored logs"""
    query = ds_client.query(kind='TaskLog')
    query.order = ['-timestamp']  # Show newest logs first
    logs = list(query.fetch(limit=100))
    
    formatted_logs = []
    for log in logs:
        log_time = log['timestamp'].strftime("%Y-%m-%d %H:%M:%S")
        formatted_logs.append(f"{log_time}: {log['message']}")
    
    return '<br>'.join(formatted_logs)

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

3. Configure app.yaml to allow Cloud Tasks access and keep instances running:

runtime: python36
automatic_scaling:
  min_instances: 1
handlers:
- url: /.*
  script: auto

4. Set up a recurring Cloud Tasks job to hit /execute-task on your desired schedule (e.g., every 5 seconds).

3. App Engine Flexible Environment (For Long-Running Processes) #

If you need a process that runs 24/7 without relying on request triggers, the App Engine Flexible Environment is a better fit—it allows long-running background processes.

Example Setup: #

app.yaml:

runtime: python
env: flex
entrypoint: gunicorn -b :$PORT main:app
runtime_config:
  python_version: 3.6
automatic_scaling:
  min_num_instances: 1

main.py:

from flask import Flask
import threading
import time
from datetime import datetime
import os

app = Flask(__name__)
task_logs = []

def persistent_task():
    while True:
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"Flex env task running at {timestamp}"
        task_logs.append(log_entry)
        
        if len(task_logs) > 100:
            task_logs.pop(0)
        
        time.sleep(5)

threading.Thread(target=persistent_task, daemon=True).start()

@app.route('/')
def hello():
    return 'Hello, world!'

@app.route('/task-logs')
def view_logs():
    return '<br>'.join(task_logs)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

Pros & Cons: #

• ✅ No need for request triggers—your task runs continuously.
• ❌ Higher cost than standard environment, since flexible instances are always running.

Pick the approach that fits your task's criticality and budget:

• Use background threads for low-impact tasks.
• Use Cloud Tasks for reliable, scheduled tasks.
• Use flexible environment for true 24/7 processes.

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