17370845950

本文介绍如何将双线程执行逻辑封装进 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 原则，又保持了多线程的灵活性，是构建可伸缩并发组件的推荐实践。