Python如何并行运行类中两个方法并实现对应触发与控制逻辑

核心思路 #

• Python标准线程没有安全的强制终止API，强行杀线程可能导致文件句柄泄漏、资源不释放等问题，所以用threading.Event作为线程终止的标志位，旧线程检测到标志位变更后主动安全退出
• 类内部维护当前运行的readLog线程实例，切换日志路径时直接替换实例即可
• 两个方法各自跑独立的子线程，并行运行互不阻塞

完整代码示例 #

import threading
import time
import os
from typing import Callable

class LogMonitor:
    def __init__(self, target_keyword: str, log_dir: str, callback: Callable):
        self.target_keyword = target_keyword  # 要检索的指定字符串
        self.log_dir = log_dir  # 日志文件存放目录
        self.callback = callback  # 检索到关键词后调用的指定方法
        self.current_log_path = None  # 当前正在读取的日志文件路径
        self._stop_read_event = threading.Event()  # readLog线程的终止标志
        self._current_read_thread: threading.Thread | None = None
        self._check_thread: threading.Thread | None = None

    def readLog(self, log_path: str):
        """读取日志文件，检索指定关键词"""
        self._stop_read_event.clear()
        with open(log_path, 'r', encoding='utf-8') as f:
            # 不需要读取历史日志的话保留下面这行，要读全量日志直接删除即可
            f.seek(0, os.SEEK_END)
            while not self._stop_read_event.is_set():
                line = f.readline()
                if not line:
                    time.sleep(0.1)  # 无新内容时休眠降低CPU占用
                    continue
                if self.target_keyword in line:
                    self.callback(line)  # 检索到关键词后调用指定方法

    def checkLog(self, check_interval: int = 5):
        """检测是否有新的日志文件生成，按修改时间判断最新文件"""
        last_latest_log = None
        while True:
            # 取目录下所有日志文件，后缀匹配规则可根据实际调整
            log_files = [f for f in os.listdir(self.log_dir) if f.endswith('.log')]
            if not log_files:
                time.sleep(check_interval)
                continue
            # 按修改时间倒序取最新文件
            log_files.sort(key=lambda x: os.path.getmtime(os.path.join(self.log_dir, x)), reverse=True)
            current_latest_log = os.path.join(self.log_dir, log_files[0])
            # 检测到新的日志文件
            if current_latest_log != last_latest_log:
                last_latest_log = current_latest_log
                # 终止当前正在运行的readLog线程
                if self._current_read_thread and self._current_read_thread.is_alive():
                    self._stop_read_event.set()
                    self._current_read_thread.join()  # 等待旧线程安全退出
                # 启动新的readLog线程
                self.current_log_path = current_latest_log
                self._current_read_thread = threading.Thread(target=self.readLog, args=(current_latest_log,), daemon=True)
                self._current_read_thread.start()
            time.sleep(check_interval)

    def start(self):
        """启动所有监控线程"""
        self._check_thread = threading.Thread(target=self.checkLog, daemon=True)
        self._check_thread.start()

# 测试用例
if __name__ == "__main__":
    # 测试用的回调方法，可替换为你自己的业务逻辑
    def test_callback(matched_line: str):
        print(f"检索到目标关键词，匹配行：{matched_line.strip()}")

    monitor = LogMonitor(
        target_keyword="ERROR",
        log_dir="./logs",  # 替换为实际的日志存放目录
        callback=test_callback
    )
    monitor.start()
    # 保持主线程运行
    while True:
        time.sleep(1)

关键说明 #

• 所有子线程默认设为daemon守护线程，主线程退出时会自动终止所有子线程，需要优雅退出的话可以自行增加全局终止标志位
• 日志检测的间隔默认是5秒，可根据需求调整checkLog的check_interval参数
• 示例默认跳过了启动监控前的历史日志，需要全量扫描的话删除readLog方法里的f.seek(0, os.SEEK_END)行即可

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