Python多进程/asyncio实现：健康检查文件循环更新至任务结束

原代码的核心问题在于time.sleep(interval)会让进程进入不可中断的睡眠状态，即便done_event被触发，也要等满5分钟睡眠结束后才会检查事件状态，无法及时终止健康检查循环。此外，代码未处理文件操作的异常，且对于这类IO密集型任务，使用进程的开销远大于线程。

修正后的多进程实现 #

将time.sleep替换为done_event.wait(interval)，这样当事件被触发时，wait会立即返回，无需等待完整的5分钟：

import multiprocessing
import time

def update_healthcheck(done_event):
    interval = 5 * 60  # 5分钟间隔（单位：秒）
    while not done_event.is_set():
        try:
            with open("healthcheck.txt", "w") as f:
                f.write("Health is okay.")
        except IOError as e:
            print(f"写入健康检查文件失败: {e}")
        # 等待interval秒，事件触发时会立即返回
        done_event.wait(interval)

def second_task(done_event):
    time.sleep(20)  # 模拟任务执行
    done_event.set()

if __name__ == "__main__":
    done_event = multiprocessing.Event()
    
    healthcheck_process = multiprocessing.Process(target=update_healthcheck, args=(done_event,))
    healthcheck_process.start()
    
    second_task(done_event)
    
    healthcheck_process.join()
    print("所有任务已完成。")

更优的多线程实现 #

由于健康检查和任务执行都是IO密集型操作，使用线程开销更小、效率更高：

import threading
import time

def update_healthcheck(done_event):
    interval = 5 * 60
    while not done_event.is_set():
        try:
            with open("healthcheck.txt", "w") as f:
                f.write("Health is okay.")
        except IOError as e:
            print(f"写入健康检查文件失败: {e}")
        done_event.wait(interval)

def second_task(done_event):
    time.sleep(20)
    done_event.set()

if __name__ == "__main__":
    done_event = threading.Event()
    
    healthcheck_thread = threading.Thread(target=update_healthcheck, args=(done_event,))
    healthcheck_thread.start()
    
    second_task(done_event)
    
    healthcheck_thread.join()
    print("所有任务已完成。")

额外优化建议 #

• 若需要更详细的健康状态记录，可以写入时间戳：f.write(f"Health is okay. Last checked: {time.ctime()}")
• 可以给健康检查线程设置daemon=True，这样即使主进程意外退出，线程也会随之终止（根据业务需求选择）
• 对于长时间运行的任务，建议使用日志模块替代print，方便后续排查问题

内容的提问来源于stack exchange，提问作者MaxU - stand with Ukraine