Python如何导入类实例而不重新初始化，解决多文件调用实例不一致问题

问题背景 #

需要复用同一个ClientService类实例，分别在常驻update循环和30秒间隔的apscheduler定时任务中调用。

实例初始化代码 #

from .services import ClientService
from flask import Flask

app = Flask(__name__)
client = ClientService() # 需全局复用的实例

from . import scheduler

ClientService类定义 #

class ClientService(object):
    def __init__(self):
        self.processing_demos = False
        self.demos_to_be_processed = []

    # 其余逻辑省略
    def push_demos(self, demo_list):
        self.processing_demos = True
        for demo in demo_list[::-1]:
            self.demos_to_be_processed.append(demo)

    def update(self):
        # 其余逻辑省略

两处实例调用逻辑 #

1. 多进程运行的update循环

from multiprocessing import Process, Value
from . import app, client
import time

def client_update_loop():
    while True:
        time.sleep(0.01)
        client.update()

if __name__ == "__main__":
    proc = Process(target=client_update_loop, args=())
    proc.start()  
    app.run(debug=True, use_reloader=False)
    proc.join()

2. apscheduler定时任务调用

from . import client

# 其余逻辑省略
client.push_demos(request_demos_list)

问题表现 #

两处逻辑拿到的client实例不是同一个对象，实例属性数据无法同步。尝试通过环境变量判断实例是否已创建、迁移初始化逻辑到其他文件，均无法解决问题。

根因定位 #

Python创建新进程时会完整拷贝父进程的所有对象，子进程中的client是独立副本，和主进程中定时任务、Flask服务调用的client分属不同内存空间，因此数据不会同步。

解决方案 #

方案1：替换多进程为asyncio异步实现（已验证可行） #

将原多进程update循环改为asyncio异步任务，全程在同一进程内运行，可直接共用同一个client实例，同时保证Flask服务和update循环并行运行：

import asyncio
import time
from . import client

async def client_update_loop():
    while True:
        # 若update方法包含IO操作，建议替换为asyncio.sleep(0.01)提升效率
        time.sleep(0.01)
        client.update()

# 启动异步循环
asyncio.run(client_update_loop())

其他可选方案（适用于必须保留多进程架构的场景） #

• 用multiprocessing模块提供的Manager、Value、Array等共享内存对象存储ClientService的所有可变属性，所有进程统一读写共享内存数据
• 引入Redis等独立中间件存储共享数据，所有进程通过读写中间件实现数据同步
• 改为多线程架构：Python线程共享同一进程的内存空间，可直接使用同一个实例，只需加锁避免竞态条件即可

内容的提问来源于stack exchange，提问作者Yaroslav Biloshytskyi