建站
咨詢
售后
建站咨詢
新聞中心
如果你在Flask中啟動(dòng)過(guò)子線程，然后在子線程中讀寫(xiě)過(guò)g?對(duì)象或者嘗試從request?對(duì)象中讀取url參數(shù)，那么，你肯定對(duì)下面這個(gè)報(bào)錯(cuò)不陌生：RuntimeError: Working outside of request context..
創(chuàng)新互聯(lián)公司專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù)，包含不限于網(wǎng)站設(shè)計(jì)、成都網(wǎng)站設(shè)計(jì)、潯陽(yáng)網(wǎng)絡(luò)推廣、小程序定制開(kāi)發(fā)、潯陽(yáng)網(wǎng)絡(luò)營(yíng)銷、潯陽(yáng)企業(yè)策劃、潯陽(yáng)品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運(yùn)營(yíng)等，從售前售中售后，我們都將竭誠(chéng)為您服務(wù)，您的肯定，是我們最大的嘉獎(jiǎng)；創(chuàng)新互聯(lián)公司為所有大學(xué)生創(chuàng)業(yè)者提供潯陽(yáng)建站搭建服務(wù)，24小時(shí)服務(wù)熱線：18980820575，官方網(wǎng)址：www.cdcxhl.com
例如下面這段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成功！'}
請(qǐng)求/start_thread接口就會(huì)報(bào)錯(cuò)，如下圖所示：
如果你在網(wǎng)上搜索flask thread RuntimeError: Working outside of request context. ，那么你可能會(huì)看到官方文檔或者StackOverFlow上面提供了一個(gè)裝飾器@copy_current_request_context。如下圖所示：
照著它這樣寫(xiě)，確實(shí)能解決問(wèn)題，如下圖所示：
但無(wú)論是官網(wǎng)還是StackOverFlow，它的例子都非常簡(jiǎn)單。但是我們知道，啟動(dòng)線程有很多種方法，例如：
# 方法一，啟動(dòng)簡(jiǎn)單線程
import threading
job = threading.Thread(target=函數(shù)名, args=(參數(shù)1, 參數(shù)2), kwargs={'參數(shù)3': xxx, '參數(shù)4': yyy})
job.start()
# 方法2，使用類定義線程
import threading
class Job(threading.Thread):
def __init__(self, 參數(shù)):
super().__init__()
def run(self):
print('子線程開(kāi)始運(yùn)行')
job = Job(參數(shù))
job.start()
# 方法3，使用線程池
from multiprocessing.dummy import Pool
pool = Pool(5) # 5個(gè)線程的線程池
pool.map(函數(shù)名, 參數(shù)列表)
網(wǎng)上的方法只能解決第一種寫(xiě)法的問(wèn)題。如果想使用方法2和方法3啟動(dòng)子線程，代碼應(yīng)該怎么寫(xiě)呢？如果在子線程中又啟動(dòng)子線程，再用一次@copy_current_request_context還行嗎？
相信我，你在網(wǎng)上搜索一下午，只有兩種結(jié)果：一是找不到答案，二是找到的答案是晚于2023年1月14日的，因?yàn)槭莿e人看了我這篇文章以后，再寫(xiě)的。
解答上面的問(wèn)題前，還是說(shuō)明一下我對(duì)于在后端啟動(dòng)子線程這個(gè)行為的觀點(diǎn)。例如有些人喜歡在后端掛一個(gè)爬蟲(chóng)，請(qǐng)求接口以后，通過(guò)線程啟動(dòng)爬蟲(chóng)，爬蟲(chóng)開(kāi)始爬數(shù)據(jù)。又或者，有些人在后端上面掛了一些復(fù)雜的程序代碼，請(qǐng)求接口以后，后端啟動(dòng)子線程，在子線程中運(yùn)行這些代碼。
我一向是不建議在后端又啟動(dòng)子線程去做復(fù)雜操作的。無(wú)論你使用的是Flask還是Django還是FastAPI。正確的做法應(yīng)該是使用消息隊(duì)列，后端只是把觸發(fā)任務(wù)的相關(guān)參數(shù)發(fā)送到消息隊(duì)列中。下游真正的運(yùn)行程序從消息隊(duì)列讀取到觸發(fā)參數(shù)以后，開(kāi)始運(yùn)行。
但有時(shí)候，你可能綜合考慮性價(jià)比，覺(jué)得再增加一個(gè)消息隊(duì)列，成本太高；或者干脆是要趕工期，不得不先暫時(shí)使用多線程來(lái)解決問(wèn)題，那么這篇文章將會(huì)極大幫助到你。
盡量不要在子線程中讀取請(qǐng)求相關(guān)的參數(shù)
如果你的子線程不需要讀寫(xiě)g對(duì)象，也不需要從請(qǐng)求中讀取各種參數(shù)，那么你就可以關(guān)閉這篇文章了。因?yàn)槟愕淖泳€程可以直接運(yùn)行，不會(huì)遇到什么的問(wèn)題，例如：
所以最好的解決方法，就是在啟動(dòng)子線程之前，提前先獲取到子線程需要的每一個(gè)參數(shù)，然后把這些參數(shù)在啟動(dòng)子線程的時(shí)候作為函數(shù)參數(shù)傳進(jìn)去。如果你是從零開(kāi)始寫(xiě)代碼，那么一開(kāi)始這樣做，就可以幫你避免很多麻煩。
但如果你是修改已有的代碼，并且嵌套太深，已經(jīng)沒(méi)有辦法一層一層傳入?yún)?shù)，或者代碼量太大，不知道哪些地方悄悄調(diào)用了g對(duì)象或者讀寫(xiě)了請(qǐng)求上下文，那么你可以繼續(xù)往下看。
裝飾閉包函數(shù)而不是一級(jí)函數(shù)
上面的簡(jiǎn)單多線程寫(xiě)法，有一個(gè)地方需要特別注意，被@copy_current_request_context裝飾的子線程入口函數(shù)inner_func，必須是閉包函數(shù)，不能是一級(jí)函數(shù)。如下圖所示：
如果不小心裝飾了一級(jí)函數(shù)，就會(huì)報(bào)如下的錯(cuò)誤：
線程池復(fù)制請(qǐng)求上下文
當(dāng)我們使用multiprocessing.dummy來(lái)實(shí)現(xiàn)線程池時(shí)，代碼如下：
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'開(kāi)始爬?。簕url}')
doc_id_list = [123, 456, 789, 111, 222, 333, 444]
pool = Pool(3)
pool.map(crawl, doc_id_list)
return {'success': True, 'msg': '爬取文章成功！'}
運(yùn)行效果如下圖所示：
寫(xiě)法上整體跟threading.Thread啟動(dòng)簡(jiǎn)單線程的方法差不多。
用類定義線程時(shí)復(fù)制請(qǐng)求上下文
當(dāng)我們額外定義了一個(gè)線程類時(shí)，需要把被裝飾的閉包函數(shù)傳入到子線程中，然后在子線程的run()方法中運(yùn)行：
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': '讀取文章成功！'}
運(yùn)行效果如下圖所示：
嵌套子線程復(fù)制請(qǐng)求上下文
有時(shí)候，我們先創(chuàng)建了一個(gè)子線程，然后在子線程中，又需要?jiǎng)?chuàng)建孫線程。并且在孫線程中讀取請(qǐng)求上下文。例如下面的代碼：
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'開(kāi)始爬?。簕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': '讀取文章成功！'}
此時(shí)使用@copy_current_request_context就會(huì)報(bào)您一個(gè)錯(cuò)誤：ValueError:
這個(gè)時(shí)候，我們就需要額外再創(chuàng)建一個(gè)裝飾器：
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這個(gè)裝飾器需要放到孫線程里面@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'開(kāi)始爬?。簕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': '讀取文章成功！'}
運(yùn)行效果如下圖所示，孫線程也正常啟動(dòng)了：
總結(jié)
- 非必要不在后端中創(chuàng)建子線程
- 創(chuàng)建子線程時(shí)，如果能把參數(shù)從外面?zhèn)魅?，就不要讓子線程自己去Flask的上下文讀取
- @copy_current_request_context需要裝飾閉包函數(shù)，不能裝飾一級(jí)函數(shù)
- 嵌套子線程需要同時(shí)使用@copy_current_app_context和@copy_current_request_context兩個(gè)裝飾器來(lái)裝飾孫線程的閉包函數(shù)
網(wǎng)站名稱：Flask嵌套啟動(dòng)子線程如何讀取請(qǐng)求上下文？
網(wǎng)站鏈接：http://m.fisionsoft.com.cn/article/cocpppj.html
