本文介绍如何将多线程逻辑完全封装在 python 类中，使每个对象实例自主管理其内部的两个并发线程，避免在主程序中手动创建和调度线程，提升代码复用性与可维护性。
在面向对象设计中，理想的多线程封装应遵循“职责内聚”原则：对象自身负责其并发行为的生命周期（启动、等待、清理），而非由外部协调。上述需求可通过在类中定义 run_threads() 和 join_threads() 方法实现——前者启动两个独立线程分别执行 func_one 和 func_two，后者阻塞主线程直至本实例所有子线程完成。
以下是完整、可运行的封装示例（含模拟任务与日志输出，便于验证并发行为）：
import threading import time class MyObject: def __init__(self, item_id): self.item_id = item_id # 可选：为线程安全预留属性（如 Event、lock 等） self._threads = [] def func_one(self): print(f"[{self.item_id}] func_one started") time.sleep(1.5) # 模拟耗时操作 print(f"[{self.item_id}] func_one completed") def func_two(self): print(f"[{self.item_id}] func_two started") time.sleep(1.0) print(f"[{self.item_id}] func_two completed") def run_threads(self): """启动本实例的两个工作线程""" thread1 = threading.Thread(target=self.func_one, name=f"{self.item_id}-func1") thread2 = threading.Thread(target=self.func_two, name=f"{self.item_id}-func2") # 存储引用以供后续 join（也可直接用实例属性，但列表更易扩展） self._threads = [thread1, thread2] thread1.start() thread2.start() def join_threads(self): """等待本实例所有工作线程结束""" for t in self._threads: t.join() # 使用示例：创建多个独立实例，并发运行 if __name__ == "__main__": obj1 = MyObject("item-01") obj2 = MyObject("item-02") # 启动所有实例的线程（非阻塞） obj1.run_threads() obj2.run_threads() # 主线程等待全部完成（顺序无关，自动同步） obj1.join_threads() obj2.join_threads() print("All instances finished.")✅ 关键优势：
- ✅ 高内聚：线程创建、启动、等待均在类内部完成；
- ✅ 实例隔离：每个 MyObject 实例拥有独立线程上下文，互不干扰；
- ✅ 可扩展性强：若需增加第三线程或动态控制，只需修改 run_threads() 内部逻辑；
- ✅ 资源可控：通过 _threads 列表统一管理，便于调试、监控或中断（如添加 is_alive() 检查）。
⚠️ 注意事项：
- 避免在 __init__ 中直接启动线程（易导致未完成初始化即执行）；
- 若线程需访问共享状态，请显式使用 threading.Lock 或 threading.Event 保证安全；
- join() 必须在 start() 之后调用，否则会引发 RuntimeError；
- 生产环境建议为线程命名（name= 参数），便于日志追踪与调试。
通过这种封装方式，你不仅能写出更清晰、更易测试的并发代码，也为后续集成线程池、异步任务队列等高级模式打下坚实基础。