重写threading.Thread的join方法是否存在风险？

我有一个运行无限循环执行任务的线程，希望在通过Ctrl-C退出时能正确完成资源清理。网上找到的方案较为复杂，我自行实现了如下代码且运行正常，但未找到类似实现，特此询问：重写join方法是否存在危险？

import threading
import time

class Logger(threading.Thread):
    def __init__(self) -> None:
        super().__init__()
        self.shutdown = False
    
    def run(self):
        while not self.shutdown:
            time.sleep(1)
            print("I am busy")
        self.cleanup()
    
    def cleanup(self):
        print("cleaning up")
    
    def join(self, timeout=None):
        self.shutdown = True
        return super().join(timeout=timeout)
    

if __name__ == "__main__":
    my_logger = Logger()
    my_logger.start()

    try:
        while True:
            time.sleep(5)
            print("Outside loop")
    except KeyboardInterrupt as e:
        my_logger.join()

重写join()方法确实存在一些潜在风险和不符合代码直觉的问题，主要包括：

• 违背API设计语义：join()的原生作用只是等待线程终止，你给它附加了"触发线程停止"的额外功能。其他开发者看到join()调用时，只会认为是等待操作，不会料到它还会主动终止线程，容易引发理解偏差和后续维护bug。
• 多次调用的隐患：如果不小心多次调用join()，每次都会重复设置self.shutdown = True。当前场景下可能没影响，但在复杂线程逻辑中，可能和其他控制线程状态的代码产生冲突。
• 版本兼容性风险：如果未来Python对threading.Thread的join()实现做了变更，你的重写逻辑可能会和新的原生行为不兼容，导致意外问题。

更合理的替代方案 #

保持join()的原生语义，单独提供一个触发停止的方法，让代码意图更清晰：

import threading
import time

class Logger(threading.Thread):
    def __init__(self) -> None:
        super().__init__()
        self.shutdown = False
    
    def run(self):
        while not self.shutdown:
            time.sleep(1)
            print("I am busy")
        self.cleanup()
    
    def cleanup(self):
        print("cleaning up")
    
    def shutdown_thread(self):
        self.shutdown = True

if __name__ == "__main__":
    my_logger = Logger()
    my_logger.start()

    try:
        while True:
            time.sleep(5)
            print("Outside loop")
    except KeyboardInterrupt as e:
        my_logger.shutdown_thread()
        my_logger.join()

你的当前实现虽然能正常运行，但从代码可读性、可维护性角度并不推荐。遵循API原生语义，分离"触发停止"和"等待结束"的逻辑，能让代码更清晰易懂，减少潜在问题。

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