Python3：如何在While True循环中仅重启已结束线程？解决迭代报错

我需要让三个不同线程运行test1函数，每个线程的执行耗时不同。线程执行完成时解锁，启动时创建重置锁，仅在线程重置时解锁。需求是线程启动后快速循环执行，只重启已完成的线程，不干扰未完成的线程，直到所有线程执行完毕。锁路径需根据操作系统/用户修改。尝试过创建新字典、更新当前字典等方法均无效，现在遇到RuntimeError: dictionary keys changed during iteration错误，代码及报错信息如下：

原代码 #

import threading
import time
from quicklock3 import singleton
import os

def test1(duration, message, name):
    time.sleep(duration)
    os.remove('/Users/engineering/.lock/%s.lock' % name)#Unlock Thread
    print(message+'----------------------------------------------------------------FINISHED')

ServiceDict = {
    'T1':threading.Thread(daemon=True, target=test1, args=(20,"T1 FINISHED", "T1")),
    'T2':threading.Thread(daemon=True, target=test1, args=(5,"T2 FINISHED", "T2")),
    'T3':threading.Thread(daemon=True, target=test1, args=(1,"T3 FINISHED", "T3"))
}

for Service in ServiceDict:
    try:
        os.remove('/Users/engineering/.lock/%s.lock' % Service)
    except Exception as e:
        print(e)
    try:
        os.remove('/Users/engineering/.lock/%s_reset.lock' % Service)
    except Exception as e:
        print(e)
        
time.sleep(5)
        
while True:
    
    for Service in ServiceDict:
        print(Service)
        time.sleep(2)
        if os.path.exists('/Users/engineering/.lock/%s_reset.lock' % Service):
            print("NO THREAD UPDATE FOR "+Service)
        else:
            print("UPDATING DICTIONARY FOR "+Service)
            ServiceDict[Service] = ServiceDict.pop(Service)
            print(ServiceDict)
            time.sleep(2)
            
    for Service in ServiceDict:
        if os.path.exists('/Users/engineering/.lock/%s_reset.lock' % Service):
            print('RESET ALREADY LOCKED...'+Service)
            time.sleep(1)
            continue
        else:
            singleton(Service) #Lock Thread
            print('LOCKING THREAD... '+Service)
            singleton(Service+'_reset') #Lock Reset
            print("STARTING THREAD... "+Service)
            ServiceDict[Service].start()
            time.sleep(1)
        
    for Service in ServiceDict:
        if os.path.exists('/Users/engineering/.lock/%s.lock' % Service):
            print("SCRIPT STILL RUNNNING... "+Service)
            time.sleep(1)
            continue
        else:
            print("RESETTING THREAD... "+Service+' ++++++++++++++++++++++++++++++++++++++++++++++++++++RESETTING')
            ServiceDict[Service].join()
            os.remove('/Users/engineering/.lock/%s_reset.lock' % Service)#Unlock Reset
            time.sleep(1)
            
    print("LOOP FINISHED") 
    time.sleep(1)

报错信息 #

RuntimeError                              Traceback (most recent call last)
<ipython-input-4-449e94d77d30> in <module>
     31     time.sleep(5)
     32 
---> 33     for Service in ServiceDict:
     34         print(Service)
     35         time.sleep(2)

错误原因 #

这个错误是因为在遍历字典ServiceDict的过程中修改了字典的键集合（比如使用pop操作再重新赋值），Python不允许在迭代字典时改变它的结构，直接触发了RuntimeError。另外原代码中还有一个隐藏问题：线程对象启动后不能重复调用start()，必须创建新的线程实例才能重启。

解决方案及修改后的代码 #

import threading
import time
from quicklock3 import singleton
import os

# 请根据你的操作系统/用户路径修改锁目录
LOCK_DIR = '/Users/engineering/.lock/'
# 确保锁目录存在，避免创建锁文件时出错
os.makedirs(LOCK_DIR, exist_ok=True)

def test1(duration, message, name):
    time.sleep(duration)
    # 解锁线程：删除对应锁文件
    os.remove(os.path.join(LOCK_DIR, f'{name}.lock'))
    print(f'{message}----------------------------------------------------------------FINISHED')

# 封装线程创建函数，方便每次重置时生成新的线程实例
def create_thread(name, duration, msg):
    return threading.Thread(daemon=True, target=test1, args=(duration, msg, name))

# 初始化线程字典
ServiceDict = {
    'T1': create_thread('T1', 20, "T1 FINISHED"),
    'T2': create_thread('T2', 5, "T2 FINISHED"),
    'T3': create_thread('T3', 1, "T3 FINISHED")
}

# 清除残留的锁文件
for service in ServiceDict:
    lock_path = os.path.join(LOCK_DIR, f'{service}.lock')
    reset_lock_path = os.path.join(LOCK_DIR, f'{service}_reset.lock')
    try:
        os.remove(lock_path)
    except Exception as e:
        print(e)
    try:
        os.remove(reset_lock_path)
    except Exception as e:
        print(e)

time.sleep(5)

while True:
    # 遍历字典的键副本（转成列表），避免迭代时修改字典结构触发错误
    for service in list(ServiceDict.keys()):
        print(service)
        time.sleep(2)
        reset_lock_path = os.path.join(LOCK_DIR, f'{service}_reset.lock')
        if os.path.exists(reset_lock_path):
            print(f"NO THREAD UPDATE FOR {service}")
        else:
            print(f"UPDATING DICTIONARY FOR {service}")
            # 生成新的线程实例替换旧的，因为旧线程无法重复启动
            ServiceDict[service] = create_thread(service, 
                                                 20 if service == 'T1' else 5 if service == 'T2' else 1,
                                                 f"{service} FINISHED")
            print(ServiceDict)
            time.sleep(2)
    
    for service in list(ServiceDict.keys()):
        reset_lock_path = os.path.join(LOCK_DIR, f'{service}_reset.lock')
        if os.path.exists(reset_lock_path):
            print(f'RESET ALREADY LOCKED...{service}')
            time.sleep(1)
            continue
        else:
            singleton(service)  # 锁定线程
            print(f'LOCKING THREAD... {service}')
            singleton(f'{service}_reset')  # 锁定重置
            print(f"STARTING THREAD... {service}")
            ServiceDict[service].start()
            time.sleep(1)
    
    for service in list(ServiceDict.keys()):
        lock_path = os.path.join(LOCK_DIR, f'{service}.lock')
        if os.path.exists(lock_path):
            print(f"SCRIPT STILL RUNNNING... {service}")
            time.sleep(1)
            continue
        else:
            print(f"RESETTING THREAD... {service} ++++++++++++++++++++++++++++++++++++++++++++++++++++RESETTING")
            ServiceDict[service].join()
            reset_lock_path = os.path.join(LOCK_DIR, f'{service}_reset.lock')
            os.remove(reset_lock_path)  # 解锁重置
            time.sleep(1)
    
    print("LOOP FINISHED") 
    time.sleep(1)

关键修改点 #

1. 遍历字典副本：用list(ServiceDict.keys())生成固定的键列表进行遍历，即使原字典结构变化也不会影响迭代过程，解决RuntimeError。
2. 线程实例重建：封装create_thread函数，每次重置线程时生成新的Thread实例，解决旧线程无法重复启动的问题。
3. 锁路径优化：用os.path.join统一管理锁路径，添加目录存在检查，避免硬编码路径错误。
4. 移除无效操作：删除了无意义的ServiceDict[Service] = ServiceDict.pop(Service)操作，改为直接创建新线程实例。

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