前言

一般web后台服务基本要求的三高要求高并发，高性能，高可用中对于我们的高并发高性能中一般需要处理几个点：

• 快速响应

所以如果我们的业务处理中存在一些长耗时的任务的话，一般的处理措施就是需要进行一步话。避免前端长时间等待。

通常的几个方案：

• 多线程（开启一个新的线程来进行耗时任务的执行）
• 多进程（开启一个或多个进程来进行耗时任务的执行）
• Celery 消息队列

1 多线程（使用线程池）

使用线程池的几个好处说明：

• 把IO资源调用进行并发处理（如网络请求，文件操作，数据库访问，远程API访问）
• 规避无线扩张的创建线程数引起的过度资源消耗，管控线程的创建和重用。

示例代码：

if __name__ == '__main__':
    app = create_app()
    print(app.url_map)

    from concurrent.futures import ThreadPoolExecutor

    # 线程池所能同时进行的最大数量
    pool_executor = ThreadPoolExecutor(max_workers=20)

    # 定义需要一部处理的任务
    def run_task_1(msg='长时间耗时任务01处理'):
        print(msg)
        time.sleep(10)
        print("任务01处理-耗时任务执行结束")

        # 定义需要一部处理的任务


    def run_task_2(msg='长时间耗时任务02处理'):
        print(msg)
        time.sleep(10)
        print("任务02处理-耗时任务执行结束")

    @app.route('/executor')
    def test():
        # 交给线程去处理耗时任务
        pool_executor.submit(run_task_1)
        pool_executor.submit(run_task_2)
        return "异步任务接口已提交后台线程执行！"


    app.run(host='127.0.0.1', port='8808', debug=True,threaded=False)

启动访问我们的接口：http://127.0.0.1:8808/executor

查看控制台的输出：

 * Running on http://127.0.0.1:8808/ (Press CTRL+C to quit)
MiddlewareTraceLogger-----before_request
2021-03-01 11:42:23.941 | INFO     | common.ext.trace_logger:trace_add_log_record:50 - {"traceid": "e5b12e6c577b4d0984fc43200723bbd6", "trace_index": 1, "event_des": "请求开始", "msg_dict": {"headers": {"X-Access-Token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJ4aWFvemhvbmd0b25neHVlIiwiZXhwIjoxNjE0NDExMTE4LCJpYXQiOjE2MTQyMzgzMTgsImF1ZCI6Ind3dy54aWFvemhvbmcuY29tIiwic3ViIjoienl4Iiwic2NvcGVzIjpbIm9wZW4iXSwiZGF0YSI6eyJ1c2VybmFtZSI6InhpYW96aG9uZyIsInVzZXJpZCI6IjEifX0.lDSBGm6houMka35uS_JFBbQYs1EYn4EpBjygfUS0cLU", "User-Agent": "PostmanRuntime/7.26.8", "Accept": "*/*", "Cache-Control": "no-cache", "Postman-Token": "d062d9dd-5839-4c52-ab71-cde02c97a169", "Host": "127.0.0.1:8808", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive"}, "url": "http://127.0.0.1:8808/executor", "method": "GET", "ip": "127.0.0.1", "params": {"args": "", "form": "", "json": ""}, "start_time": "2021-03-01 11:42:23.940692"}}
2021-03-01 11:42:23.947 | INFO     | common.ext.trace_logger:trace_add_log_record:50 - {"traceid": "e5b12e6c577b4d0984fc43200723bbd6", "trace_index": 2, "event_des": "请求结束", "msg_dict": {"cost_time": "15.0ms", "rsp_data": "异步任务接口已提交后台线程执行！"}}
127.0.0.1 - - [01/Mar/2021 11:42:23] "GET /executor HTTP/1.1" 200 -
长时间耗时任务01处理
长时间耗时任务02处理MiddlewareTraceLogger-----after_request=
# 十秒后才输出下面的结果
sssssssssss 200 OK
任务01处理-耗时任务执行结束
任务02处理-耗时任务执行结束

PS:上面的服务的运行app.run(host='127.0.0.1', port='8808', debug=True)默认的是开启多了多线程处理请求。

如果需要开启单线程需要显示的设置关闭线程：如下代码(后续用于对比单线程下的异步非阻塞)：app.run(host='127.0.0.1', port='8808', debug=True,threaded=False)

if __name__ == '__main__':
    app = create_app()
    print(app.url_map)

    from concurrent.futures import ThreadPoolExecutor

    # 线程池所能同时进行的最大数量
    executor = ThreadPoolExecutor(max_workers=20)

    # 定义需要一部处理的任务
    def run(msg='长时间耗时任务处理'):
        print(msg)
        time.sleep(10)
        print("耗时任务执行结束")


    @app.route('/xiancheng')
    def xiancheng():
        # 交给线程去处理耗时任务
        # executor.submit(run)

        # 打印出当前线程的名称和id
        pid = os.getpid()
        print("当前主进程pid：" +str(pid))
        import threading
        print("当前线程名称："+threading.currentThread().name)
        print("当前线程名称："+str(threading.currentThread().ident))
        return '线程显示'


    app.run(host='127.0.0.1', port='8808', debug=True,threaded=False)

2 多进程（最大进程数即为电脑CPU的核数最佳）

为什么引入多进程？

• 由于GIL锁的原因，CPU密集型运行的任务,在多个线程只能用一个cpu，这时多进程的执行效率要比多线程高。
• 所以多进程一般适用于CPU密集型运行的任务
• 关于多进程中因为每个进程的开启后，都说分别运行各自的进程里面，相关环境是隔离的

if __name__ == '__main__':
    app = create_app()
    print(app.url_map)

    from multiprocessing import cpu_count
    from concurrent.futures import ProcessPoolExecutor

    #使用CPU核数梳理为最佳
    pool_process = ProcessPoolExecutor(max_workers=cpu_count())
    # 斐波拉切-属于CPU密集型运算-任务
    def run_task_fib(n):
        if n == 1:
            return [1]
        if n == 2:
            return [1, 1]
        fibs = [1, 1]
        for i in range(2, n):
            fibs.append(fibs[-1] + fibs[-2])
        print("结果：",fibs)
        return fibs

    @app.route('/executor')
    def test():
        # 交给线程去处理耗时任务
        rusult =pool_process.map(run_task_fib(20))
        print(rusult)
        import json
        return "OK"


    # run_task_fib(20)
    app.run(host='127.0.0.1', port='8808', debug=True,threaded=False)

访问：http://127.0.0.1:8808/executor 查看控制台输出结果：

MiddlewareTraceLogger-----before_request
结果： [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987, 1597, 2584, 4181, 6765]
<generator object _chain_from_iterable_of_lists at 0x0000021750421DE0>
MiddlewareTraceLogger-----after_request=
sssssssssss 200 OK
2021-03-01 12:12:08.993 | INFO     | common.ext.trace_logger:trace_add_log_record:50 - {"traceid": "5450fcf132eb466296f92e1e01c3233d", "trace_index": 1, "event_des": "请求开始", "msg_dict": {"headers": {"X-Access-Token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJ4aWFvemhvbmd0b25neHVlIiwiZXhwIjoxNjE0NDExMTE4LCJpYXQiOjE2MTQyMzgzMTgsImF1ZCI6Ind3dy54aWFvemhvbmcuY29tIiwic3ViIjoienl4Iiwic2NvcGVzIjpbIm9wZW4iXSwiZGF0YSI6eyJ1c2VybmFtZSI6InhpYW96aG9uZyIsInVzZXJpZCI6IjEifX0.lDSBGm6houMka35uS_JFBbQYs1EYn4EpBjygfUS0cLU", "User-Agent": "PostmanRuntime/7.26.8", "Accept": "*/*", "Cache-Control": "no-cache", "Postman-Token": "c81ca691-bd94-411d-abb6-6aeab6d985de", "Host": "127.0.0.1:8808", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive"}, "url": "http://127.0.0.1:8808/executor", "method": "GET", "ip": "127.0.0.1", "params": {"args": "", "form": "", "json": ""}, "start_time": "2021-03-01 12:12:08.992828"}}
2021-03-01 12:12:08.994 | INFO     | common.ext.trace_logger:trace_add_log_record:50 - {"traceid": "5450fcf132eb466296f92e1e01c3233d", "trace_index": 2, "event_des": "请求结束", "msg_dict": {"cost_time": "5.0ms", "rsp_data": "OK"}}
127.0.0.1 - - [01/Mar/2021 12:12:08] "GET /executor HTTP/1.1" 200 -

3 简单实践-引入芹菜 Celery 结合redis做消息中间件做异步任务

下面的这种整合Flask方式比较麻烦，其实可以单独的分开不一定整合一起。看个人业务需要吧！

3.1 库安装依赖（不建议windos环境下完整高版本的）

• Celery 5.0.5
• redis 3.5.3

如果是windows下的建议修改为：

• Celery 3.1.18
• redis 2.XXX

定义celery对象（基于windows下的Celery 5.0.5 和redis 3.5.3 ）

具体的异步任务和对象创建

from celery import Celery
import time

app = Celery('异步任务尝试', backend='redis://127.0.0.1:6379/6', broker='redis://127.0.0.1:6379/5')

# 定义异步任务函数
@app.task
def add(x, y):
    print("异步任务开始------>")
    time.sleep(5)
    print("异步任务结束--------------")
    return x + y


if __name__ == "__main__":
    app.start()

启动我们的工作者：

1）python+文件名的方式启动方式（基于windows下的Celery 5.0.5 和redis 3.5.3 ）

(AntFlask) D:\code\python\local_python\AntFlask\backend\asyntasks>python test.py worker --loglevel=info

2）celery -A + 模块名（Celery下app对应所在的模块的名称）（基于windows下的Celery 5.0.5 和redis 3.5.3 ）

# (AntFlask) D:\code\python\local_python\AntFlask\backend\asyntasks>celery -A test worker --loglevel=info

输出结果：

(AntFlask) D:\code\python\local_python\AntFlask\backend\asyntasks>python test.py worker --loglevel=info

 -------------- celery@DESKTOP-16CKEN1 v5.0.5 (singularity)
--- ***** -----
-- ******* ---- Windows-10-10.0.18362-SP0 2021-03-01 13:54:10
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         异步任务尝试:0x20f925be160
- ** ---------- .> transport:   redis://127.0.0.1:6379/5
- ** ---------- .> results:     redis://127.0.0.1:6379/6
- *** --- * --- .> concurrency: 6 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . 异步任务尝试.add

[2021-03-01 13:54:10,530: INFO/MainProcess] Connected to redis://127.0.0.1:6379/5
[2021-03-01 13:54:10,537: INFO/MainProcess] mingle: searching for neighbors
[2021-03-01 13:54:10,911: INFO/SpawnPoolWorker-1] child process 46632 calling self.run()
[2021-03-01 13:54:10,947: INFO/SpawnPoolWorker-2] child process 24052 calling self.run()
[2021-03-01 13:54:11,008: INFO/SpawnPoolWorker-3] child process 24996 calling self.run()
[2021-03-01 13:54:11,122: INFO/SpawnPoolWorker-4] child process 10832 calling self.run()
[2021-03-01 13:54:11,156: INFO/SpawnPoolWorker-5] child process 47884 calling self.run()
[2021-03-01 13:54:11,248: INFO/SpawnPoolWorker-6] child process 44828 calling self.run()
[2021-03-01 13:54:11,559: INFO/MainProcess] mingle: all alone
[2021-03-01 13:54:11,567: INFO/MainProcess] celery@DESKTOP-16CKEN1 ready.

然后在我们的上面的接口查实引入我们的异步任务

if __name__ == '__main__':
    app = create_app()
    print(app.url_map)

    @app.route('/executor')
    def test():
        # 引入异步的处理任务
        from backend.asyntasks.test import add
        add.delay(1, 3)
        return "OK"


    # run_task_fib(20)
    app.run(host='127.0.0.1', port='8808', debug=True,threaded=False)

启动服务，访问http://127.0.0.1:8808/executor

查看我们的异步任务信息：出现了异常：

[2021-03-01 14:09:33,400: ERROR/MainProcess] Received unregistered task of type 'backend.asyntasks.test.add'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
b'[[1, 3], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]' (81b)
Traceback (most recent call last):
  File "C:\Users\mayn\.virtualenvs\AntFlask\lib\site-packages\celery\worker\consumer\consumer.py", line 555, in on_task_received
    strategy = strategies[type_]
KeyError: 'backend.asyntasks.test.add'

查看任务是已经添加到了列表里面：

查看我们的后端redis存储的列表数据：

坑爹，难道还是版本不支持的问题：切换linux 下运行是OK! 所以还是不建议再windos是测试使用Celery!

End

纯属个人以往实践中相关经验之谈，如有纰漏，还希望大佬们不吝赐教，多多指点！

小钟同学 | 文  【原创】| QQ：308711822