Python：Flask使用ThreadPoolExecutor執行異步任務

測試代碼

# -*- coding: utf-8 -*-

import time
from concurrent.futures import ThreadPoolExecutor

from flask import Flask, request

executor = ThreadPoolExecutor()

app = Flask(__name__)


# 模拟耗時任務
def run_job(name):
    time.sleep(5)
    print('run_job complete', name)


@app.route('/task')
def run_task():
    """
    同步執行
    http://127.0.0.1:5000/task
    """
    name = request.args.get('name')

    run_job(name=name)

    return {'ret': 'ok'}


@app.route('/async_task')
def run_async_task():
    """
    異步執行
    http://127.0.0.1:5000/async_task
    """
    name = request.args.get('name')

    executor.submit(run_job, name=name)

    return {'ret': 'ok'}


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

apache2-utils壓力測試工具測試：

# 使用說明：
$ ab -n 請求數 -c 并發數  URL


# 同步執行任務
$ ab -n 100 -c 10 http://127.0.0.1:5000/task

Time taken for tests:   55.100 seconds


# 異步執行任務
$ ab -n 100 -c 10 http://127.0.0.1:5000/async_task

Time taken for tests:   0.097 seconds