Python并行任务隔离：使用进程而非线程解决共享变量问题

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

# 模拟一个数据库模块，其中包含一个全局变量
class DB:
    DB_MODE = 1 # 默认值，表示某种数据库模式

def find_request_with_thread(flag=False, task_id=None):
    """
    此函数将在线程中执行，并尝试修改DB.DB_MODE。
    """
    print(f"Task {task_id} (Thread ID: {asyncio.current_task()._asyncio_task_identity[0]}) - Initial DB_MODE: {DB.DB_MODE}, Flag: {flag}")
    if flag:
        DB.DB_MODE = 0 # 线程共享此变量，修改会影响其他线程
        print(f"Task {task_id} - DB_MODE changed to 0 by this thread.")
    else:
        print(f"Task {task_id} - DB_MODE remains {DB.DB_MODE}.")

    time.sleep(0.05) # 模拟一些工作
    return {"task_id": task_id, "final_db_mode": DB.DB_MODE, "flag_applied": flag}

async def main_with_threads():
    version_required = [True, False, True, False]
    tasks = []

    print(f"Main process initial DB_MODE: {DB.DB_MODE}")

    # 使用ThreadPoolExecutor来运行线程任务
    with ThreadPoolExecutor(max_workers=4) as executor:
        loop = asyncio.get_running_loop()

        for i, request_flag in enumerate(version_required):
            tasks.append(loop.run_in_executor(
                executor,
                find_request_with_thread,
                request_flag,
                i
            ))

        processed_data = await asyncio.gather(*tasks)

    print("\n--- All threaded tasks completed ---")
    for result in processed_data:
        print(f"Task result: {result}")

    print(f"\nMain process final DB_MODE: {DB.DB_MODE}")

if __name__ == "__main__":
    print("--- Running with ThreadPoolExecutor ---")
    asyncio.run(main_with_threads())

import asyncio
from concurrent.futures import ProcessPoolExecutor # 关键修改：使用ProcessPoolExecutor
import os
import time

# 模拟一个数据库模块，其中包含一个全局变量
class DB:
    DB_MODE = 1 # 默认值

def find_request_with_process(flag=False, task_id=None):
    """
    此函数将在独立的进程中执行。
    每个进程都会获得DB.DB_MODE的独立副本，修改不会影响其他进程。
    """
    current_pid = os.getpid()
    # 打印当前进程的初始DB_MODE
    print(f"Process {current_pid} (Task: {task_id}) - Initial DB_MODE: {DB.DB_MODE}, Flag: {flag}")

    if flag:
        DB.DB_MODE = 0 # 此修改仅对当前进程有效，不会影响其他进程或主进程
        print(f"Process {current_pid} (Task: {task_id}) - DB_MODE changed to 0 for this process.")
    else:
        print(f"Process {current_pid} (Task: {task_id}) - DB_MODE remains {DB.DB_MODE}.")

    time.sleep(0.05) # 模拟一些工作
    # 返回此进程的最终DB_MODE
    return {"task_id": task_id, "process_id": current_pid, "final_db_mode": DB.DB_MODE, "flag_applied": flag}

async def main_with_processes():
    version_required = [True, False, True, False]
    tasks = []

    print(f"Main process (PID: {os.getpid()}) initial DB_MODE: {DB.DB_MODE}")

    # 使用ProcessPoolExecutor来运行进程任务
    # max_workers 通常设置为CPU核心数
    with ProcessPoolExecutor(max_workers=os.cpu_count() or 4) as executor:
        loop = asyncio.get_running_loop()

        for i, request_flag in enumerate(version_required):
            tasks.append(loop.run_in_executor(
                executor,
                find_request_with_process, # 函数在独立进程中执行
                request_flag,
                i
            ))

        processed_data = await asyncio.gather(*tasks)

    print("\n--- All processed tasks completed ---")
    for result in processed_data:
        print(f"Task result: {result}")

    # 验证主进程的DB.DB_MODE是否未受子进程影响
    print(f"\nMain process (PID: {os.getpid()}) final DB_MODE: {DB.DB_MODE}")

if __name__ == "__main__":
    print("--- Running with ProcessPoolExecutor ---")
    asyncio.run(main_with_processes())