基于WSGI部署的Python Flask应用如何指定单个Worker执行任务？

针对你提到的需要单Worker执行的任务（清理过期文件、同步数据到Redis、删除文本特定行），可以用以下几种实用方案实现：

方案一：文件互斥锁（跨Worker进程同步） #

利用文件锁实现进程间互斥，保证同一时间只有一个Worker能拿到锁执行任务，其他Worker会直接跳过。适合按需触发的任务（比如接口调用触发清理）。

代码示例（Unix/Linux环境）： #

import fcntl
import os
from functools import wraps

def single_process_lock(lock_file_path='/tmp/flask_task_lock.lock'):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            lock_file = None
            try:
                # 打开锁文件，不存在则创建
                lock_file = open(lock_file_path, 'w')
                # 获取排他锁，非阻塞模式
                fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
                # 拿到锁，执行任务
                return func(*args, **kwargs)
            except BlockingIOError:
                # 其他进程已持有锁，直接返回
                return "Task is already running by another worker"
            finally:
                if lock_file:
                    # 释放锁并关闭文件
                    fcntl.flock(lock_file, fcntl.LOCK_UN)
                    lock_file.close()
        return wrapper
    return decorator

# 使用装饰器标记需要单进程执行的任务
@single_process_lock()
def clean_expired_files():
    # 逻辑：遍历目录，删除过期文件
    pass

@single_process_lock()
def sync_to_redis():
    # 逻辑：读取文件数据，同步到Redis
    pass

Windows环境可以替换fcntl为win32file模块的锁机制，核心逻辑一致。

方案二：指定特定Worker执行定时任务 #

如果用Gunicorn这类WSGI服务器，可以通过Worker的ID标识，只让第一个Worker（ID=0）启动定时任务调度器（比如APScheduler），其他Worker不执行。适合周期性执行的任务。

代码示例： #

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import os

app = Flask(__name__)

def clean_expired_files():
    # 清理过期文件逻辑
    pass

def sync_to_redis():
    # 同步到Redis逻辑
    pass

def delete_specific_lines():
    # 删除文本特定行逻辑
    pass

# 仅在Worker ID为0时启动调度器
if os.environ.get('GUNICORN_WORKER_ID') == '0':
    scheduler = BackgroundScheduler()
    # 添加定时任务，比如每小时执行一次清理
    scheduler.add_job(clean_expired_files, 'interval', hours=1)
    scheduler.add_job(sync_to_redis, 'interval', minutes=30)
    scheduler.add_job(delete_specific_lines, 'interval', hours=2)
    scheduler.start()

if __name__ == '__main__':
    app.run()

启动Gunicorn时，Worker ID环境变量会被自动设置，无需额外配置。

方案三：独立进程执行任务（最可靠） #

把这些单实例任务完全从WSGI Worker中剥离出来，写成独立的Python脚本，用系统定时任务（Linux的crontab、Windows的任务计划程序）定期执行。这种方案完全不受Worker数量、重启的影响，是最稳定的方式。

示例步骤： #

1. 写独立脚本task_worker.py：

# task_worker.py
def clean_expired_files():
    # 清理逻辑
    pass

def sync_to_redis():
    # 同步逻辑
    pass

def delete_specific_lines():
    # 删除行逻辑
    pass

if __name__ == '__main__':
    clean_expired_files()
    sync_to_redis()
    delete_specific_lines()

2. 设置Linux crontab，每小时执行一次：
   打开crontab编辑：crontab -e
   添加一行：

0 * * * * /usr/bin/python3 /path/to/task_worker.py

这样系统会每小时自动启动一个独立进程执行所有单实例任务，完全和WSGI Worker隔离。

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