本文介绍如何将多线程逻辑完全封装在 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.")✅ 关键优势：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- ✅ 高内聚：线程创建、启动、等待均在类内部完成；
- ✅ 实例隔离：每个 MyObject 实例拥有独立线程上下文，互不干扰；
- ✅ 可扩展性强：若需增加第三线程或动态控制，只需修改 run_threads() 内部逻辑；
- ✅ 资源可控：通过 _threads 列表统一管理，便于调试、监控或中断（如添加 is_alive() 检查）。
⚠️ 注意事项：
- 避免在 __init__ 中直接启动线程（易导致未完成初始化即执行）；
- 若线程需访问共享状态，请显式使用 threading.Lock 或 threading.Event 保证安全；
- join() 必须在 start() 之后调用，否则会引发 RuntimeError；
- 生产环境建议为线程命名（name= 参数），便于日志追踪与调试。
通过这种封装方式，你不仅能写出更清晰、更易测试的并发代码，也为后续集成线程池、异步任务队列等高级模式打下坚实基础。
