为什么我的递归Python代码中堆大小与线程数持续增长？

你的代码出现内存泄漏和线程累积的核心原因有两个：

1. 非守护线程的资源未及时回收
   你创建的线程默认是非守护线程（daemon=False），这类线程结束后，Python解释器会保留其退出状态信息，直到显式调用join()方法。由于你没有保存线程引用也未执行join()，这些线程对象无法被及时垃圾回收（GC），导致线程数在监测工具中持续累积，表现为“空”线程（已终止但未被回收的线程对象）。

2. Work实例被线程绑定方法持续持有
   线程的target是self.by_duration——这是一个绑定方法，会持有Work实例的引用。在with块结束后，Work实例本应被销毁，但由于线程仍在运行（或线程对象未被回收），实例的引用计数始终大于0，无法被GC回收。每次循环都会创建新的Work实例和线程对象，未被回收的对象持续占用堆内存，导致堆大小线性增长。

方案1：使用守护线程自动回收资源 #

将线程设置为守护线程，这样线程结束后，Python会自动回收其资源，无需显式join()：

import time
import threading

class Work:

    def by_duration(self, seconds: float, block: bool = True) -> None:
        if seconds < 0:
            raise ValueError("seconds >= 0")

        if seconds == 0:
            return

        if block:
            time.sleep(seconds)
        else:
            # 设置daemon=True，将线程标记为守护线程
            t = threading.Thread(target=self.by_duration, args=(seconds, True), daemon=True)
            t.start()
            return

    def clean_up(self):
        pass

    def __enter__(self):
        return self

    def __exit__(self, *args) -> None:
        self.clean_up()


while True:
    with Work() as p:
        p.by_duration(3, block=False)
        time.sleep(3)

方案2：显式管理线程生命周期（更可靠） #

在Work类中保存线程引用，在clean_up中调用join()等待线程结束，确保实例和线程资源被完全回收：

import time
import threading

class Work:
    def __init__(self):
        self._threads = []

    def by_duration(self, seconds: float, block: bool = True) -> None:
        if seconds < 0:
            raise ValueError("seconds >= 0")

        if seconds == 0:
            return

        if block:
            time.sleep(seconds)
        else:
            t = threading.Thread(target=self.by_duration, args=(seconds, True))
            t.start()
            self._threads.append(t)
            return

    def clean_up(self):
        # 等待所有线程结束，释放资源
        for thread in self._threads:
            thread.join()
        self._threads.clear()

    def __enter__(self):
        return self

    def __exit__(self, *args) -> None:
        self.clean_up()


while True:
    with Work() as p:
        p.by_duration(3, block=False)
        time.sleep(3)

内容的提问来源于stack exchange，提问作者Cam.Davidson.Pilon