如何在multiprocess场景下正确删除导入时定义变量绑定的对象

问题根因 #

你遇到的问题本质是全局单例的引用没有被主动释放：

• __del__方法仅在实例的引用计数归0时才会被Python解释器触发执行
• 你用全局变量PRODUCER_SINGLETON持有了Producer实例的引用，子进程退出时，CPython的默认销毁流程不会主动清空全局作用域的变量引用，因此实例的引用计数始终不为0，__del__自然不会执行
• 多进程场景下，子进程的全局变量是独立于主进程的副本，主进程也无法干预子进程内该变量的引用计数

解决方案 #

方案1：手动清理全局引用 #

最简单的方式是在子进程任务执行结束前，主动将全局单例变量置为None，释放对实例的引用：

from multiprocessing import Process

PRODUCER_SINGLETON = None

def func_test():
    Producer.create_producer()
    # 任务结束后主动清理全局引用
    global PRODUCER_SINGLETON
    PRODUCER_SINGLETON = None


class Producer:
    @classmethod
    def create_producer(cls):
        global PRODUCER_SINGLETON
        if PRODUCER_SINGLETON is None:
            PRODUCER_SINGLETON = Producer()
        return PRODUCER_SINGLETON

    def __init__(self):
        print("got into produce")
        self.produce = "toto"

    def __del__(self):
        print("got into delete")


if __name__ == "__main__":
    process = Process(
        target=func_test,
    )
    process.start()
    process.join()

运行后可以看到got into delete正常输出。

方案2：注册进程退出钩子自动清理（业务代码无侵入） #

如果不想在每个业务函数里手动加清理逻辑，可以用Python标准库的atexit模块注册进程退出回调，子进程正常退出时会自动执行清理逻辑：

from multiprocessing import Process
import atexit

PRODUCER_SINGLETON = None

def _cleanup_producer():
    global PRODUCER_SINGLETON
    PRODUCER_SINGLETON = None

class Producer:
    @classmethod
    def create_producer(cls):
        global PRODUCER_SINGLETON
        if PRODUCER_SINGLETON is None:
            PRODUCER_SINGLETON = Producer()
            # 仅第一次创建实例时注册退出钩子
            atexit.register(_cleanup_producer)
        return PRODUCER_SINGLETON

    def __init__(self):
        print("got into produce")
        self.produce = "toto"

    def __del__(self):
        print("got into delete")

def func_test():
    Producer.create_producer()

if __name__ == "__main__":
    process = Process(
        target=func_test,
    )
    process.start()
    process.join()

这种方式不需要修改任何业务侧的调用代码，只要调用过create_producer，进程退出时就会自动清理引用。

方案3：用上下文管理器管理生命周期（更安全） #

如果单例的使用范围是固定的代码块，推荐用上下文管理器来控制生命周期，即使代码块内抛出异常也能保证实例被正确销毁：

from multiprocessing import Process

PRODUCER_SINGLETON = None

class Producer:
    def __init__(self):
        print("got into produce")
        self.produce = "toto"

    def __del__(self):
        print("got into delete")

    @classmethod
    def get_instance(cls):
        global PRODUCER_SINGLETON
        if PRODUCER_SINGLETON is None:
            PRODUCER_SINGLETON = Producer()
        return PRODUCER_SINGLETON

# 上下文管理器
class ProducerContext:
    def __enter__(self):
        return Producer.get_instance()

    def __exit__(self, exc_type, exc_val, exc_tb):
        global PRODUCER_SINGLETON
        PRODUCER_SINGLETON = None

def func_test():
    with ProducerContext() as producer:
        # 在这里执行你的业务逻辑
        print(f"using producer: {producer.produce}")

if __name__ == "__main__":
    process = Process(
        target=func_test,
    )
    process.start()
    process.join()

注意事项 #

如果子进程是被系统信号强制杀死（比如SIGKILL、未捕获的SIGTERM），所有Python层的清理逻辑都不会执行，__del__也不会触发，这种场景需要额外编写信号处理逻辑。

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