Python多线程：实现线程间依赖执行（指定线程完成后启动）

问题描述 #

现有如下Python线程创建代码：

for i in range (5):
        thread = threading.Thread(target=some_fun,args=("thread : {}".format(i), fun_args))
        thread.start()
        
thread6 = threading.Thread(target=some_fun,args=("thread : 6", fun_args))
thread6.start()
thread7 = threading.Thread(target=some_fun,args=("thread : 7", fun_args))
thread7.start()

上述代码动态创建了5个线程（编号0-4，对应需求中的1-5），并手动创建了线程6、7，所有线程执行同一函数some_fun，仅参数不同。

需求：

• 线程6需在线程1（对应代码中i=0）和线程2（对应代码中i=1）执行完成后立即启动
• 线程7需在线程4（对应代码中i=3）执行完成后立即启动
• 线程6无需等待线程3、4、5、7完成；线程7仅需等待线程4，无需等待其他线程
• 需支持扩展更多类似的线程依赖关系

解决方案 #

方案1：使用join()方法（简单依赖场景）

核心思路是保存需要等待的线程对象，在启动依赖线程前调用这些线程的join()方法，阻塞直到目标线程完成。

修改后的代码示例：

import threading

def some_fun(thread_name, args):
    # 模拟线程执行任务，替换为实际业务逻辑
    print(f"{thread_name} 开始执行")
    import time
    time.sleep(2)
    print(f"{thread_name} 执行完成")

fun_args = "some_args"

# 保存线程对象，方便后续依赖等待
threads = []
for i in range(5):
    thread = threading.Thread(target=some_fun, args=(f"thread : {i+1}", fun_args))  # 调整编号为1-5
    threads.append(thread)
    thread.start()

# 等待线程1和线程2完成后启动线程6
threads[0].join()  # 对应thread : 1
threads[1].join()  # 对应thread : 2
thread6 = threading.Thread(target=some_fun, args=("thread : 6", fun_args))
thread6.start()

# 等待线程4完成后启动线程7
threads[3].join()  # 对应thread : 4
thread7 = threading.Thread(target=some_fun, args=("thread : 7", fun_args))
thread7.start()

# 可选：等待所有线程完成
for thread in threads + [thread6, thread7]:
    thread.join()

方案2：使用threading.Event（灵活依赖场景）

当依赖关系更复杂（比如多个线程等待同一个线程完成，或者一个线程等待多个非连续线程），使用Event更灵活。每个线程执行完成后触发对应的事件，依赖线程等待事件触发后再执行。

代码示例：

import threading

def some_fun(thread_name, args, finish_event=None):
    print(f"{thread_name} 开始执行")
    import time
    time.sleep(2)
    print(f"{thread_name} 执行完成")
    # 线程完成后触发事件
    if finish_event:
        finish_event.set()

fun_args = "some_args"

# 为每个需要被依赖的线程创建完成事件
thread1_finish = threading.Event()
thread2_finish = threading.Event()
thread4_finish = threading.Event()

threads = []
# 启动线程1-5，并绑定对应的完成事件
for i in range(5):
    thread_name = f"thread : {i+1}"
    finish_event = None
    if i+1 == 1:
        finish_event = thread1_finish
    elif i+1 == 2:
        finish_event = thread2_finish
    elif i+1 == 4:
        finish_event = thread4_finish
    thread = threading.Thread(target=some_fun, args=(thread_name, fun_args, finish_event))
    threads.append(thread)
    thread.start()

# 线程6等待thread1和thread2的事件触发后启动
def start_thread6():
    thread1_finish.wait()
    thread2_finish.wait()
    some_fun("thread : 6", fun_args)

thread6 = threading.Thread(target=start_thread6)
thread6.start()

# 线程7等待thread4的事件触发后启动
def start_thread7():
    thread4_finish.wait()
    some_fun("thread : 7", fun_args)

thread7 = threading.Thread(target=start_thread7)
thread7.start()

# 可选：等待所有线程完成
for thread in threads + [thread6, thread7]:
    thread.join()

方案3：封装依赖管理器（复杂依赖扩展场景）

如果后续有大量线程和复杂依赖关系，可以封装一个简单的管理器，统一管理线程的完成状态和依赖启动逻辑：

import threading
from collections import defaultdict

class ThreadDependencyManager:
    def __init__(self):
        self.finish_events = defaultdict(threading.Event)
        self.threads = []

    def add_thread(self, thread_name, target, args, depends_on=None):
        """
        添加线程，支持指定依赖的线程名称列表
        :param thread_name: 线程唯一标识名称
        :param target: 线程执行函数
        :param args: 函数参数
        :param depends_on: 依赖的线程名称列表，如["thread : 1", "thread : 2"]
        """
        def wrapped_target():
            # 等待所有依赖线程完成
            if depends_on:
                for dep_name in depends_on:
                    self.finish_events[dep_name].wait()
            # 执行目标函数
            target(*args)
            # 触发当前线程的完成事件
            self.finish_events[thread_name].set()

        thread = threading.Thread(target=wrapped_target, name=thread_name)
        self.threads.append(thread)
        return thread

# 使用示例
manager = ThreadDependencyManager()

fun_args = "some_args"

# 添加线程1-5，无依赖
for i in range(5):
    thread_name = f"thread : {i+1}"
    manager.add_thread(thread_name, some_fun, (thread_name, fun_args)).start()

# 添加线程6，依赖thread1和thread2
manager.add_thread("thread : 6", some_fun, ("thread : 6", fun_args), depends_on=["thread : 1", "thread : 2"]).start()

# 添加线程7，依赖thread4
manager.add_thread("thread : 7", some_fun, ("thread : 7", fun_args), depends_on=["thread : 4"]).start()

# 等待所有线程完成
for thread in manager.threads:
    thread.join()

总结 #

• 简单依赖场景优先用join()，代码直观易维护
• 需要灵活处理多对多依赖时用threading.Event
• 大量线程或复杂依赖推荐封装管理器，便于后续扩展

内容的提问来源于stack exchange，提问作者Mohammad Sunny