如何将线程封装在类内并运行该类的多个实例？

完全可以把线程的创建、启动和等待逻辑封装到类内部，让每个实例独立管理自己的线程生命周期，这样代码更整洁、内聚性更强，也避免了外部大量重复的线程操作代码。

实现示例 #

以下是修改后的代码，将线程逻辑完全封装在MyObject类中：

import threading

class MyObject:
    def __init__(self, item_id):
        self.item_id = item_id
        # 初始化时创建线程对象，绑定当前实例的方法
        self._threads = [
            threading.Thread(target=self.func_one),
            threading.Thread(target=self.func_two)
        ]

    def func_one(self):
        print(f"实例 {self.item_id} 的func_one运行中")
        # 替换为你的业务逻辑

    def func_two(self):
        print(f"实例 {self.item_id} 的func_two运行中")
        # 替换为你的业务逻辑

    def start(self):
        """启动当前实例的所有线程"""
        for thread in self._threads:
            thread.start()

    def wait(self):
        """等待当前实例的所有线程执行完毕"""
        for thread in self._threads:
            thread.join()

# 使用方式
obj1 = MyObject("item-01")
obj2 = MyObject("item-02")

# 启动所有实例的线程
obj1.start()
obj2.start()

# 等待所有线程结束
obj1.wait()
obj2.wait()

关键说明 #

• 线程对象在类的__init__方法中创建，绑定实例的func_one和func_two方法，threading.Thread会自动处理实例方法所需的self参数。
• 通过start()和wait()方法对外暴露线程操作接口，外部代码无需关心内部线程的创建细节。
• 如果需要支持重复启动线程（注意：单个线程对象只能启动一次），可以修改start()方法，每次调用时重新创建线程对象：

def start(self):
    """每次调用都重新创建并启动线程"""
    self._threads = [
        threading.Thread(target=self.func_one),
        threading.Thread(target=self.func_two)
    ]
    for thread in self._threads:
        thread.start()

这种封装方式让类的职责更清晰，后续扩展线程相关功能（比如添加线程状态监控、优雅停止线程等）也更方便。

内容的提问来源于stack exchange，提问作者R.Orteza