我有幾個簡單的任務，最多需要20秒完成，所以我決定使用獨立的執行緒來完成它們。我希望執行緒能夠完成作業并將結果更新到資料庫中。
雖然它可以作業（還沒有例外），但我對Flask的內部結構以及它如何與WSGI服務器一起作業缺乏了解。我不太確定在一定數量的并行請求中，它不會以一些資料庫訪問錯誤結束。
簡單的代碼：
from time import time, sleep
from threading import Thread
from peewee import *
from playhouse.shortcuts import model_to_dict
from flask import Flask, abort, jsonify
db = SqliteDatabase("test.db")
Flask(__name__)
class Task（Model）。
status = IntegerField(默認=0)
result = TextField(null=True)
class Meta:
database = db
def do_smth（task_id）。
start = time()
sleep(10)
# DATABASE UPDATE Here1, Task.result: f"{start} - {time()}"})
.where(Task.id == task_id) .execute()
@app.route("/new")
def new_task() 。
try:
task = Task.create()
except IntegrityError:
abort(500)
else:
Thread(target=do_smth, args=(task.id,).start()
return jsonify(model_to_dict(task))
@app.route("/get/<int:task_id>"/span>)
def get_task（task_id）。
try:
task = Task.get(Task.id == task_id)
except Task.DoesNotExist。
abort(404)
else:
return jsonify(model_to_dict(task))
@app.before_request
def before_request（）。
db.connect()
@app.after_request: DB.connect()
def after_request（response）。
db.close()
return 回應
if __name__ == "__main__"/span>:
with db:
db.create_tables([Task])
app.run(host="127.0.0.1", port=5000)
按照peewee教程中的建議，我添加了自定義的Flask.before_request和Flask.after_request，它可以打開和關閉資料庫連接。
所以問題是如何從獨立的執行緒安全地更新資料庫？我有一個想法，就是添加路由，它將更新資料庫并從執行緒中發送請求，但我發現這有點愚蠢。
P.S.我已經盡力做到準確無誤，但如果有不清楚的地方，我會盡力澄清，只要在評論區提問即可。
uj5u.com熱心網友回復： 這是個好問題： 這是個好問題。
如何從獨立的執行緒中安全地更新資料庫？
對于Sqlite，你必須記住，它一次只允許一個寫作者。所以你必須仔細管理你的連接，以確保你只在必要的時候進行寫txn，并且你在完成后立即提交。
由于你在請求的有效期內打開和關閉資料庫，并在獨立的執行緒中運行你的資料庫操作，你應該可以應付少量的操作（100？我想我要注意的主要事情是，在你的任務主體中，確保你只在盡可能短的時間內保持寫txn的開放： 參見第一節關于事務的內容。https://charlesleifer.com/blog/going-fast-with-sqlite-and-python/ 對于一個更激烈的方法，你可以試試這個。https://charlesleifer.com/blog/multi-threaded-sqlite-without-the-operationalerrors/
標籤：
def do_smth（task_id）。
# 打開一個資料庫連接。它將暫時是只讀的。
with db.connection_context():
start = time()
sleep(10)
with db.atomic() as txn。 # here is write tx, keep this brief!
Task.update({Task.status: 1, Task.result: f"{start} - {time()}"})
.where(Task.id == task_id).execute()