Flask中创建多线程和多进程

创建多进程

import multiprocessing
from applications.create_app import create_app
from applications.models.app.app import AppData
from common.model.base_model import db
from sqlalchemy.orm.attributes import flag_modified


app = create_app()


def worker(record_id, new_id):
    with app.app_context():
        # 读取数据并加锁
        # record = db.session.query(AppData).filter_by(id=record_id).with_for_update().first()
        record = db.session.query(AppData).filter(AppData.id == record_id).first()

        # 更新数据
        if not record.data.get("updated_ids"):
            record.data["updated_ids"] = []
        record.data["updated_ids"].append(new_id)
        flag_modified(record, "data")

        # 提交事务
        db.session.commit()


def test_thread():

    # 创建多个进程，模拟多个worker同时更新同一条数据
    processes = []
    for i in range(1, 9):
        p = multiprocessing.Process(target=worker, args=(1, i))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    # 检查更新结果
    with app.app_context():
        record = db.session.query(AppData).filter(AppData.id == 1).first()
        print(record.data["updated_ids"])


if __name__ == "__main__":
    test_thread()

创建多线程

def test_():
    test_thread()


def test_thread():
    # 创建两个线程，模拟两个worker同时更新同一条数据
    thread1 = threading.Thread(target=worker, args=(1, 1))
    thread2 = threading.Thread(target=worker, args=(1, 2))
    thread3 = threading.Thread(target=worker, args=(1, 3))
    thread4 = threading.Thread(target=worker, args=(1, 4))
    thread5 = threading.Thread(target=worker, args=(1, 5))
    thread6 = threading.Thread(target=worker, args=(1, 6))
    thread7 = threading.Thread(target=worker, args=(1, 7))
    thread8 = threading.Thread(target=worker, args=(1, 8))

    # 启动线程
    thread1.start()
    thread2.start()
    thread3.start()
    thread4.start()
    thread5.start()
    thread6.start()
    thread7.start()
    thread8.start()

    # 等待线程完成
    thread1.join()
    thread2.join()
    thread3.join()
    thread4.join()
    thread5.join()
    thread6.join()
    thread7.join()
    thread8.join()

    # 检查更新结果
    record = AppData.query.filter(AppData.id == 1).first()
    print(record.data["updated_ids"])  # [1, 2]


def worker(record_id, new_id):
    with db.app.app_context():
        # 读取数据并加锁
        # record = db.session.query(AppData).filter_by(id=record_id).first()
        record = db.session.query(AppData).filter_by(id=record_id).with_for_update().first()

        # 更新数据
        if not record.data.get("updated_ids"):
            record.data["updated_ids"] = []
        record.data["updated_ids"].append(new_id)
        flag_modified(record, "data")

        # 提交事务
        db.session.commit()