如何同步处理Flask请求 保障pywinauto任务单实例按接收顺序执行

你之前的方案出现并发问题的核心原因有两个：

1. 配置CELERY_ALWAYS_EAGER=True时，Celery不会把任务转发给worker进程，直接在当前请求的线程中同步执行，Flask默认开启多线程处理请求，多个请求同时进来就会同时运行pywinauto任务，导致并发
2. 关闭CELERY_ALWAYS_EAGER后如果Celery worker的并发数大于1，多个工作进程/线程会同时从RabbitMQ拉取任务执行，也会出现并发

以下是两个可直接落地的解决方案：

适合请求量较大、需要任务持久化的场景

步骤1：修改tasks.py配置 #

from celery import Celery
import os
# Windows环境专用配置，避免fork相关报错
os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '0')

app = Celery('tasks', backend='rpc://', broker='amqp://guest@localhost')
# 禁用任务预取，保证任务严格按入队顺序执行
app.conf.worker_prefetch_multiplier = 1
# 固定worker并发数为1，同一时间仅执行一个任务
app.conf.worker_concurrency = 1

@app.task
def inputClient(crmid):
    # 你的原有pywinauto执行逻辑
    uiInputClient(crmid)
    return "executed"

步骤2：正确启动Celery Worker #

Windows环境下必须使用单线程的solo池，执行以下命令启动：
celery -A tasks worker --pool=solo --loglevel=info

步骤3：调整接口路由代码 #

@app.route('/ClientEntry', methods=['GET'])
def createClient():
    crmid = request.args.get('crmid')
    if not crmid:
        raise BadRequest('CRM ID cannot be empty')
    result = inputClient.apply_async(args=[crmid])
    try:
        exec_res = result.get(timeout=120)
        print('inside ' + exec_res)
        return json.dumps({'success':True}), 200, {'ContentType':'application/json'} 
    except Exception as e:
        return json.dumps({'success':False, 'error':str(e)}), 500

如果你的请求量不大，不需要任务持久化，可以直接用全局互斥锁实现串行执行，无需引入Celery、RabbitMQ等组件：

from flask import Flask, request, json
from werkzeug.exceptions import BadRequest
import threading

app = Flask(__name__)
# 全局互斥锁，保证同一时间仅一个pywinauto任务执行
pywin_lock = threading.Lock()

# 你的原有pywinauto执行逻辑
def uiInputClient(crmid):
    pass

@app.route('/ClientEntry', methods=['GET'])
def createClient():
    crmid = request.args.get('crmid')
    if not crmid:
        raise BadRequest('CRM ID cannot be empty')
    # 加锁后执行任务，后续请求会阻塞等待前一个任务执行完释放锁
    with pywin_lock:
        try:
            uiInputClient(crmid)
            return json.dumps({'success':True}), 200, {'ContentType':'application/json'}
        except Exception as e:
            return json.dumps({'success':False, 'error':str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, threaded=True)

• pywinauto依赖桌面用户会话，运行Flask或Celery Worker的进程必须在已登录的用户桌面下启动，不能注册为系统后台服务，否则无法识别UI元素
• 若任务执行时间较长，需对应调整Web服务器的超时配置，避免请求未执行完就被断开
• 高并发场景下推荐使用Celery方案，可修改接口逻辑为先返回任务提交成功，后续通过回调或查询接口通知用户执行结果，避免请求长时间阻塞

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