本文介绍如何将双线程执行逻辑封装进 python 类中，使每个对象实例自主管理其内部的两个并发线程，避免在主程序中手动创建和管理大量线程对象。
在面向对象编程中，良好的封装性意味着对象应尽可能自治——包括其并发行为。原始写法将 threading.Thread 的创建、启动与等待（join）全部暴露在类外部，不仅破坏了封装，还导致主逻辑臃肿、可维护性差，且难以扩展（例如增加线程数或统一生命周期管理）。
通过将线程管理内聚到类中，我们可以定义清晰的接口：run_threads() 启动本实例的全部工作线程，join_threads() 等待它们完成。这样，每个 MyObject 实例都成为一个“可并发执行的单元”。
以下是优化后的完整实现：
import threading
import time
class MyObject:
def __init__(self, item_id):
self.item_id = item_id
def func_one(self):
print(f"[{self.item_id}] func_one started")
time.sleep(1.5) # 模拟耗时操作
print(f"[{self.item_id}] func_one finished")
def func_two(self):
print(f"[{self.item_id}] func_two started")
time.sleep(1.0)
print(f"[{self.item_id}] func_two finished")
def run_threads(self):
"""启动本实例的两个工作线程"""
self.thread1 = threading.Thread(target=self.func_one, name=f"{self.item_id}-func1")
self.thread2 = threading.Thread(target=self.func_two, name=f"{self.item_id}-func2")
self.thread1.start()
self.thread2.start()
def join_threads(self):
"""阻塞等待本实例所有工作线程结束"""
self.thread1.join()
self.thread2.join()
# 使用示例：创建并并发运行多个实例
obj1 = MyObject("item-01")
obj2 = MyObject("item-02")
# 启动所有实例的内部线程（非阻塞）
obj1.run_threads()
obj2.run_threads()
# 等待全部完成（顺序调用，但各实例内线程仍并行执行）
obj1.join_threads()
obj2.join_threads()
print("All instances completed.")✅ 关键优势：
- 高内聚：线程生命周期与对象绑定，self.thread1/self.thread2 属于实例状态；
- 易复用：新增实例只需 objN = MyObject(...); objN.run_threads(); objN.join_threads()；
- 可扩展：如需支持动态线程数，可将 func_one/func_two 抽象为列表或注册函数；
- 便于调试：可通过 threading.current_thread().name 或日志明确归属。
⚠️ 注意事项：
- 避免在 __init__ 中直接启动线程（可能导致对象未完全构造就执行）；
- 若需线程安全共享状态，请使用 threading.Lock 或其他同步原语；
- join_threads() 必须在 run_threads() 之后调用，否则会引发 RuntimeError（尝试 join 未启动的线程）；
- 如需异步等待或超时控制，可改用 thread.join(timeout=...) 并检查 thread.is_alive()。
这种设计既符合 OOP 原则，又保持了多线程的灵活性，是构建可伸缩并发组件的推荐实践。
