如何在独立进程中周期性调用Python实例方法

你的问题核心在于多进程间的内存隔离——子进程启动时会复制父进程的Buffer实例，子进程里调用_refresh_data()修改的是副本的data属性，主进程的原始实例完全感知不到这个变化，自然拿不到更新后的数据。

下面给出两种针对性解决方案，根据你的实际场景选择：

方案一：用线程替代进程（推荐，IO密集场景更高效） #

如果你的_refresh_data()是耗时的IO操作（比如读取文件、接口请求），线程是更优选择：IO等待时线程会释放GIL，不会阻塞主进程，而且线程共享同一内存空间，修改的就是原始实例的属性，不需要额外跨进程同步。

修改后的代码：

import threading
import random
import time


class Buffer:
    def __init__(self):
        self.lock = threading.Lock()
        # 启动自动更新线程
        self.thread = threading.Thread(target=self._auto_refresh, daemon=True)
        self.thread.start()

        # 初始数据更新
        self.data = None
        self.update()

    def _refresh_data(self):
        """模拟耗时的数据读取，子类可重写此方法"""
        numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9]
        data = [random.choice(numbers) for _ in range(3)]
        self.data = data

    def _auto_refresh(self):
        """周期性自动刷新逻辑"""
        while True:
            with self.lock:
                self._refresh_data()
            time.sleep(10)

    def update(self):
        """手动触发更新"""
        with self.lock:
            self._refresh_data()
    
    def get_data(self):
        return self.data


if __name__ == '__main__':
    buffer = Buffer()
    data_first = buffer.get_data()
    print(f"初始数据: {data_first}")
    time.sleep(11)
    data_second = buffer.get_data()
    print(f"更新后数据: {data_second}")
    print(f"数据是否更新: {data_first != data_second}")

这个方案完全满足你的三个需求：

• 保留update()方法支持手动更新
• 自动更新线程后台运行，不阻塞主进程
• 子类只需重写_refresh_data()就能复用自动更新逻辑

方案二：多进程+共享数据结构（适合CPU密集场景） #

如果你的_refresh_data()是CPU密集型操作，必须用多进程，那就用multiprocessing.Manager创建跨进程共享的数据结构，让主进程和子进程共用同一份数据：

import multiprocessing as mp
import random
import time


def _auto_refresh_helper(shared_data, lock, refresh_func):
    while True:
        with lock:
            # 调用刷新函数获取新数据，更新共享结构
            new_data = refresh_func()
            shared_data['data'] = new_data
        time.sleep(10)


class Buffer:
    def __init__(self):
        self.manager = mp.Manager()
        # 创建跨进程共享的字典存储数据
        self.shared_data = self.manager.dict()
        self.lock = mp.Lock()

        # 启动子进程，传递刷新函数和共享数据
        self.proc = mp.Process(
            target=_auto_refresh_helper,
            args=(self.shared_data, self.lock, self._refresh_data),
            daemon=True
        )
        self.proc.start()

        # 初始数据更新
        self.update()

    def _refresh_data(self):
        """模拟耗时的数据读取，子类可重写此方法"""
        numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9]
        return [random.choice(numbers) for _ in range(3)]

    def update(self):
        """手动触发更新"""
        with self.lock:
            new_data = self._refresh_data()
            self.shared_data['data'] = new_data
    
    def get_data(self):
        return self.shared_data.get('data')


if __name__ == '__main__':
    buffer = Buffer()
    data_first = buffer.get_data()
    print(f"初始数据: {data_first}")
    time.sleep(11)
    data_second = buffer.get_data()
    print(f"更新后数据: {data_second}")
    print(f"数据是否更新: {data_first != data_second}")

这个方案通过共享字典传递数据，避免了实例复制的问题，同样满足你的所有需求，适合CPU密集场景。

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