如何在不使用多重继承的前提下实现Python线程相关抽象类？

当然可以不用多重继承实现需求，这里提供两种可行的重构方案：

方案一：组合模式（委托实现） #

遵循「组合优于继承」的设计原则，把threading.Thread作为内部成员变量，将线程相关操作委托给它，自身只负责定义抽象逻辑和核心流程。

import threading
from abc import abstractmethod

class BackgroundThread:
    def __init__(self):
        self._stop_event = threading.Event()
        # 内部维护Thread实例，绑定自定义的_run方法
        self._thread = threading.Thread(target=self._run)

    def start(self):
        """启动线程"""
        self._thread.start()

    def stop(self) -> None:
        """触发线程停止事件"""
        self._stop_event.set()

    def join(self, timeout=None):
        """等待线程结束"""
        self._thread.join(timeout)

    def _stopped(self) -> bool:
        """检查线程是否需要停止"""
        return self._stop_event.is_set()

    @abstractmethod
    def startup(self) -> None:
        """线程启动前的资源初始化方法，子类必须实现"""
        raise NotImplementedError()

    @abstractmethod
    def shutdown(self) -> None:
        """线程停止后的资源清理方法，子类必须实现"""
        raise NotImplementedError()

    @abstractmethod
    def handle(self) -> None:
        """线程循环执行的业务逻辑，子类必须实现（注意不能长时间阻塞）"""
        raise NotImplementedError()

    def _run(self) -> None:
        """线程实际执行的核心流程"""
        self.startup()
        while not self._stopped():
            self.handle()
        self.shutdown()

使用示例 #

class MyTaskThread(BackgroundThread):
    def startup(self):
        print("线程初始化资源")

    def shutdown(self):
        print("线程清理资源")

    def handle(self):
        print("执行业务逻辑")
        threading.sleep(1)

# 启动线程
thread = MyTaskThread()
thread.start()

# 运行一段时间后停止
threading.sleep(5)
thread.stop()
thread.join()

这种方式完全避免了继承，耦合度更低，后续修改线程实现也更灵活。

方案二：指定元类替代多重继承 #

Python抽象基类的核心是ABCMeta元类，ABC类只是一个封装了该元类的便捷基类。我们可以直接给BackgroundThread指定ABCMeta元类，同时只继承threading.Thread，这样既保留线程功能，又能使用抽象方法特性。

import threading
from abc import abstractmethod, ABCMeta

class BackgroundThread(threading.Thread, metaclass=ABCMeta):
    def __init__(self):
        super().__init__()
        self._stop_event = threading.Event()

    def stop(self) -> None:
        self._stop_event.set()

    def _stopped(self) -> bool:
        return self._stop_event.is_set()

    @abstractmethod
    def startup(self) -> None:
        """线程启动前的初始化方法"""
        raise NotImplementedError()

    @abstractmethod
    def shutdown(self) -> None:
        """线程停止后的清理方法"""
        raise NotImplementedError()

    @abstractmethod
    def handle(self) -> None:
        """线程循环执行的业务逻辑"""
        raise NotImplementedError()

    def run(self) -> None:
        self.startup()
        while not self._stopped():
            self.handle()
        self.shutdown()

说明 #

• 仅继承threading.Thread，通过metaclass=ABCMeta开启抽象类特性，避免了多重继承ABC
• 原有代码结构几乎不用改动，子类的实现方式和之前完全一致

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