Python程序运行1-2小时后无报错冻结，寻求技术帮助

问题背景 #

程序初期运行正常，但1-2小时后会无报错、无崩溃地冻结，添加print语句也无法获取任何输出，相关代码如下：

import datetime
import os
import sys
import time
import dw2


class Watcher(object):
    file = None
    running = True
    refresh_delay_secs = 1

    # Constructor
    def __init__(self, watch_file, call_func_on_change=None, *args, **kwargs):
        self._cached_stamp = 0
        self.filename = watch_file
        self.file = watch_file
        self.call_func_on_change = call_func_on_change
        self.args = args
        self.kwargs = kwargs

    # Look for changes
    def look(self):
        stamp = os.stat(self.filename).st_mtime
        if stamp != self._cached_stamp:
            self._cached_stamp = stamp
            if self.call_func_on_change is not None:
                self.call_func_on_change(*self.args, **self.kwargs)
            return "Updated"
        else:
            return "Not Updated"

    # Keep watching in a loop
    def watch(self):
        try:
            time.sleep(self.refresh_delay_secs)
            result=self.look()
            if result.__eq__("Updated"):
                dw2.dwc(self.file)
            print(result)
        except KeyboardInterrupt:
            print('\nDone')
        except FileNotFoundError:
            pass
        except:
            print('Unhandled error: %s' % sys.exc_info()[0])

    def restart(self):
        os.system('cls')
        os.execv(sys.executable,  ['python'] + sys.argv)

watch_file = 'status.txt'

watcher = Watcher(watch_file)  
watcher = Watcher(watch_file)

while watcher.running :
    try :
        watcher.watch()  
    except:
        restart()

排查分析 #

• 第三方库阻塞：核心触发逻辑dw2.dwc(self.file)是第三方库调用，未明确超时机制，长时间运行可能因IO等待、锁竞争陷入无限阻塞，直接导致主线程冻结。
• 异常捕获过度：外层while循环使用裸except:捕获所有异常，包括线程阻塞、资源耗尽等隐性问题，掩盖了真实错误信息。
• 冗余实例化：连续两次实例化Watcher，第二次覆盖第一次实例，可能引发意外状态。
• 函数调用错误：外层except中的restart()未指定为watcher.restart()，会触发NameError，但被全局捕获后无任何提示。

解决方案 #

1. 给第三方库调用添加超时保护 #

用线程池包装dw2.dwc，避免阻塞主线程：

from concurrent.futures import ThreadPoolExecutor

# 修改Watcher类的watch方法
def watch(self):
    try:
        time.sleep(self.refresh_delay_secs)
        result = self.look()
        if result == "Updated":
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(dw2.dwc, self.file)
                # 设置30秒超时，可根据实际调整
                try:
                    future.result(timeout=30)
                except TimeoutError:
                    print("dw2.dwc执行超时，已终止")
                    executor.shutdown(wait=False)
        print(result)
    except KeyboardInterrupt:
        print('\nDone')
    except FileNotFoundError:
        pass
    except Exception as e:
        print(f'错误详情: {str(e)}')

2. 修复异常捕获逻辑 #

移除裸except:，改为捕获特定异常并打印完整栈信息：

while watcher.running :
    try :
        watcher.watch()
    except Exception as e:
        import traceback
        traceback.print_exc()  # 打印完整异常栈，定位问题
        watcher.restart()  # 正确调用实例的重启方法

3. 清理冗余代码 #

删除重复的Watcher实例化：

watch_file = 'status.txt'
watcher = Watcher(watch_file)  # 仅保留一次实例化

4. 优化文件监听逻辑 #

替换低效的轮询方式，用watchdog库实现事件驱动监听，减少资源消耗：

# 先安装依赖：pip install watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class FileChangeHandler(FileSystemEventHandler):
    def __init__(self, callback, target_file):
        self.callback = callback
        self.target_file = target_file

    def on_modified(self, event):
        if not event.is_directory and event.src_path == self.target_file:
            self.callback(self.target_file)

def start_watching(file_path):
    event_handler = FileChangeHandler(dw2.dwc, file_path)
    observer = Observer()
    observer.schedule(event_handler, path=os.path.dirname(file_path) or '.', recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

# 替换原有的Watcher逻辑
start_watching(watch_file)

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