Flask中嵌套启动子线程的方法示例详解

作者:青南 时间:2023-01-13 21:13:08 

如果你在Flask中启动过子线程,然后在子线程中读写过g对象或者尝试从request对象中读取url参数,那么,你肯定对下面这个报错不陌生:RuntimeError: Working outside of request context..

例如下面这段Flask代码:

import threading
from flask import Flask, request
app = Flask(__name__)
def inner_func():
   doc_id = request.args.get('doc_id', '')
   print(f'用户ID为:{doc_id}')
@app.route('/start_thread')
def start_thread():
   thread = threading.Thread(target=inner_func)
   thread.start()
   return {'success': True, 'msg': '获取用户ID成功!'}

请求/start_thread接口就会报错,如下图所示:

Flask中嵌套启动子线程的方法示例详解

如果你在网上搜索flask thread RuntimeError: Working outside of request context. ,那么你可能会看到官方文档或者StackOverFlow上面提供了一个装饰器@copy_current_request_context。如下图所示:

Flask中嵌套启动子线程的方法示例详解

照着它这样写,确实能解决问题,如下图所示:

Flask中嵌套启动子线程的方法示例详解

但无论是官网还是StackOverFlow,它的例子都非常简单。但是我们知道,启动线程有很多种方法,例如:

# 方法一,启动简单线程
import threading
job = threading.Thread(target=函数名, args=(参数1, 参数2), kwargs={'参数3': xxx, '参数4': yyy})
job.start()
# 方法2,使用类定义线程
import threading
class Job(threading.Thread):
   def __init__(self, 参数):
       super().__init__()
   def run(self):
       print('子线程开始运行')
job = Job(参数)
job.start()
# 方法3,使用线程池
from multiprocessing.dummy import Pool
pool = Pool(5)  # 5个线程的线程池
pool.map(函数名, 参数列表)

网上的方法只能解决第一种写法的问题。如果想使用方法2和方法3启动子线程,代码应该怎么写呢?如果在子线程中又启动子线程,再用一次@copy_current_request_context还行吗?

相信我,你在网上搜索一下午,只有两种结果:一是找不到答案,二是找到的答案是晚于2023年1月14日的,因为是别人看了我这篇文章以后,再写的。

解答上面的问题前,还是说明一下我对于在后端启动子线程这个行为的观点。例如有些人喜欢在后端挂一个爬虫,请求接口以后,通过线程启动爬虫,爬虫开始爬数据。又或者,有些人在后端上面挂了一些复杂的程序代码,请求接口以后,后端启动子线程,在子线程中运行这些代码。

我一向是不建议在后端又启动子线程去做复杂操作的。无论你使用的是Flask还是Django还是FastAPI。正确的做法应该是使用消息队列,后端只是把触发任务的相关参数发送到消息队列中。下游真正的运行程序从消息队列读取到触发参数以后,开始运行。

但有时候,你可能综合考虑性价比,觉得再增加一个消息队列,成本太高;或者干脆是要赶工期,不得不先暂时使用多线程来解决问题,那么这篇文章将会极大帮助到你。

尽量不要在子线程中读取请求相关的参数

如果你的子线程不需要读写g对象,也不需要从请求中读取各种参数,那么你就可以关闭这篇文章了。因为你的子线程可以直接运行,不会遇到什么的问题,例如:

Flask中嵌套启动子线程的方法示例详解

所以最好的解决方法,就是在启动子线程之前,提前先获取到子线程需要的每一个参数,然后把这些参数在启动子线程的时候作为函数参数传进去。如果你是从零开始写代码,那么一开始这样做,就可以帮你避免很多麻烦。

但如果你是修改已有的代码,并且嵌套太深,已经没有办法一层一层传入参数,或者代码量太大,不知道哪些地方悄悄调用了g对象或者读写了请求上下文,那么你可以继续往下看。

装饰闭包函数而不是一级函数

上面的简单多线程写法,有一个地方需要特别注意,被@copy_current_request_context装饰的子线程入口函数inner_func,必须是闭包函数,不能是一级函数。如下图所示:

Flask中嵌套启动子线程的方法示例详解

如果不小心装饰了一级函数,就会报如下的错误:

Flask中嵌套启动子线程的方法示例详解

线程池复制请求上下文

当我们使用multiprocessing.dummy来实现线程池时,代码如下:

from multiprocessing.dummy import Pool
from flask import Flask, request, copy_current_request_context, g
app = Flask(__name__)
@app.route('/start_thread', methods=['POST'])
def start_thread():
   @copy_current_request_context
   def crawl(doc_id):
       url_template = request.json.get('url_template', '')
       url = url_template.format(doc_id=doc_id)
       print(f'开始爬取:{url}')
   doc_id_list = [123, 456, 789, 111, 222, 333, 444]
   pool = Pool(3)
   pool.map(crawl, doc_id_list)
   return {'success': True, 'msg': '爬取文章成功!'}

运行效果如下图所示:

Flask中嵌套启动子线程的方法示例详解

写法上整体跟threading.Thread启动简单线程的方法差不多。

用类定义线程时复制请求上下文

当我们额外定义了一个线程类时,需要把被装饰的闭包函数传入到子线程中,然后在子线程的run()方法中运行:

import threading
from flask import Flask, request, copy_current_request_context
app = Flask(__name__)
class Job(threading.Thread):
   def __init__(self, func):
       super().__init__()
       self.func = func
   def run(self):
       self.func()
@app.route('/start_thread', methods=['POST'])
def start_thread():
   @copy_current_request_context
   def runner():
       doc_id = request.json.get('doc_id', '')
       print(f'docId的值是:{doc_id}')
   job = Job(runner)
   job.start()
   return {'success': True, 'msg': '读取文章成功!'}

运行效果如下图所示:

Flask中嵌套启动子线程的方法示例详解

嵌套子线程复制请求上下文

有时候,我们先创建了一个子线程,然后在子线程中,又需要创建孙线程。并且在孙线程中读取请求上下文。例如下面的代码:

import threading
from multiprocessing.dummy import Pool
from flask import Flask, request, copy_current_request_context
app = Flask(__name__)
def deep_func_runner(doc_id_list):
   @copy_current_request_context
   def deep_func(doc_id):
       category = request.args.get('category', '')
       url = f'https://www.kingname.info/{category}/{doc_id}'
       print(f'开始爬取:{url}')
   pool = Pool(3)
   pool.map(deep_func, doc_id_list)
@app.route('/start_thread', methods=['POST'])
def start_thread():
   @copy_current_request_context
   def runner():
       doc_id_list = [111, 222, 333, 444, 555, 666, 777, 888, 999]
       deep_func_runner(doc_id_list)
   job = threading.Thread(target=runner)
   job.start()
   return {'success': True, 'msg': '读取文章成功!'}

此时使用@copy_current_request_context就会报您一个错误:ValueError: <Token var=<ContextVar name='flask.request_ctx' at 0x103ef69a0> at 0x104446700> was created in a different Context。如下图所示:

Flask中嵌套启动子线程的方法示例详解

这个时候,我们就需要额外再创建一个装饰器:

def copy_current_app_context(f):
   from flask.globals import _app_ctx_stack
   appctx = _app_ctx_stack.top
   def _(*args, **kwargs):
       with appctx:
           return f(*args, **kwargs)
   return _

@copy_current_app_context这个装饰器需要放到孙线程里面@copy_current_request_context的上面。完整的代码为:

import threading
from multiprocessing.dummy import Pool
from flask import Flask, request, copy_current_request_context
app = Flask(__name__)
def copy_current_app_context(f):
   from flask.globals import _app_ctx_stack
   appctx = _app_ctx_stack.top
   def _(*args, **kwargs):
       with appctx:
           return f(*args, **kwargs)
   return _
def deep_func_runner(doc_id_list):
   @copy_current_app_context
   @copy_current_request_context
   def deep_func(doc_id):
       category = request.args.get('category', '')
       url = f'https://www.kingname.info/{category}/{doc_id}'
       print(f'开始爬取:{url}')
   pool = Pool(3)
   pool.map(deep_func, doc_id_list)
@app.route('/start_thread', methods=['POST'])
def start_thread():
   @copy_current_request_context
   def runner():
       doc_id_list = [111, 222, 333, 444, 555, 666, 777, 888, 999]
       deep_func_runner(doc_id_list)
   job = threading.Thread(target=runner)
   job.start()
   return {'success': True, 'msg': '读取文章成功!'}

运行效果如下图所示,孙线程也正常启动了:

Flask中嵌套启动子线程的方法示例详解

总结

  • 非必要不在后端中创建子线程

  • 创建子线程时,如果能把参数从外面传入,就不要让子线程自己去Flask的上下文读取

  • @copy_current_request_context需要装饰闭包函数,不能装饰一级函数

  • 嵌套子线程需要同时使用@copy_current_app_context@copy_current_request_context两个装饰器来装饰孙线程的闭包函数

来源:https://juejin.cn/post/7188512662920593468

标签:Flask,嵌套启动,子线程
0
投稿

猜你喜欢

  • SQL2005查看死锁存储过程sp_who_lock

    2024-01-12 20:34:13
  • 使用opencv中匹配点对的坐标提取方式

    2022-12-02 00:32:54
  • 利用Vscode进行Python开发环境配置的步骤

    2023-04-18 09:06:46
  • ASP连接access和mssql数据库的全能代码

    2008-10-12 13:17:00
  • 分享6个Go处理字符串的技巧小结

    2024-01-30 17:39:07
  • Python 2/3下处理cjk编码的zip文件的方法

    2022-08-05 17:53:40
  • python pip特殊用法之pip install -v -e .命令详解

    2022-07-02 09:56:43
  • Python 序列化 pickle/cPickle模块使用介绍

    2021-04-28 17:23:21
  • Python 中的 Counter 模块及使用详解(搞定重复计数)

    2022-08-22 15:46:17
  • vue多页面项目实现版本快照功能示例详解

    2024-04-28 09:23:25
  • 前端之vue3使用WebSocket的详细步骤

    2024-04-30 10:28:54
  • Oracle中SQL语句连接字符串的符号使用介绍

    2024-01-21 06:12:11
  • python如何实现不用装饰器实现登陆器小程序

    2022-07-02 15:52:14
  • Python3内置json模块编码解码方法详解

    2023-02-15 13:32:26
  • PyTorch模型保存与加载实例详解

    2022-02-24 20:08:56
  • python opencv进行图像拼接

    2022-09-10 16:13:01
  • Javascript简单实现面向对象编程继承实例代码

    2024-04-19 09:50:42
  • php7安装openssl扩展方法

    2023-11-14 17:34:14
  • SQL Server 数据库故障修复顶级技巧之一

    2010-05-01 18:49:00
  • 基于JS+Canvas的lucky-canvas 抽奖功能

    2024-04-19 09:51:26
  • asp之家 网络编程 m.aspxhome.com