python多线程实战——监控循环

在项目开发过程中，总结出关于多线程的知识与经验，供开发者参考

监控循环线程

监控循环是多线程最基本的一个应用之一。在特定场景下，如需要感知用户是否离线/触发保活机制时，可以创建一个后台线程，用来监控用户行为，可能的代码如下：

import time  
import threading

def monitor_loop():
    interval = 600 # 每600s监控一次
    while True:
        monitor() # 具体的监控逻辑
        time.sleep(interval)

monitro_thread = threading.thread(target = monitor_loop, daemon = True) # 守护线程  
monitro_thread.start()

一般来说，我们会设置监控线程为守护线程，因为如果进程结束了，也就没有继续监控的必要了，线程自动退出即可

监控线程控制

如果使用while True，会导致监控线程启动后无法手动关闭，需要进程结束后自动退出，灵活性不足。因此，通常通过监控器类的方式来控制监控线程，并且提供一个启动的标志，如：

import time 
import threading 

class Monitor:
    def __init__(self):
        self._started = False 
        self._lock = threading.Lock()
        self._interval = 600
        self._thread = None

    def start(self, interval=None):
        with self._lock:
            if self._started:
                return
            self._started = True
            if interval is not None:
                self._interval = interval
            self._thread = threading.Thread(target=self._run, daemon=True)
            self._thread.start()

    def stop(self):
        with self._lock:
            self._started = False

    def is_running(self):
        with self._lock:
            return self._started

    def _run(self):
        while self.is_running():
            self.monitor()
            time.sleep(interval)

    def monitor(self):
        # 具体的监控逻辑在此处实现
        print("正在监控...")

通过设置self._started和self.is_running()的方式，我们可以定义线程的运行状态，如果self._started == True，则启动监控循环后，while self.is_running()就可以运行了；而self.stop()通过设置self._started = False，来中断监控循环

由于self.start()和self.stop()（通常由主线程或其他线程调用）与监控线程内部（即_run方法）对self._started标志的读取是并发进行的，存在竞态条件。如果没有适当的同步机制，可能出现这样的问题：主线程刚刚设定self._started = False，但监控线程还没看到变化，仍然进入下一轮循环；反之亦然。
为了保证线程安全，必须用锁（如threading.Lock）将所有访问、修改self._started的操作包裹起来，实现原子性。这也是示例代码中with self._lock语句块的作用，确保任意时刻只有一个线程能够读写self._started，从而避免竞态和数据不一致的问题。

监控线程的拟同步停止

拟同步停止指的是，需要调用stop()后立即终止线程，在self.interval很小时，停止几乎是同步的(假设，self.stop()修改self._started时，while循环刚好开始新的一轮监控 + 等待，需要至少self.interval的时间，如果self.interval很小,这个影响忽略不计)，但是如果self.interval很大(如600s)，在调用self.stop()后，可能需要等待600s，不符合我们同步停止的需求。

为了解决这个问题，可以设计一个拟同步停止的工具，原理为，拆分time.sleep(long_interval)为多次time.sleep(short_interval)，每次短间隔等待后，都判断是否需要退出进程，具体实现如下：

"""
可中断的等待工具：在长间隔内按小步长睡眠并定期检查“是否继续运行”标志，
便于关闭线程时快速退出，无需等满整个配置间隔。
"""
import time
from typing import Callable


def interruptible_sleep(
    total_seconds: float,
    keep_running: Callable[[], bool],
    check_interval: float = 1.0,
) -> bool:
    """
    可中断的等待：在 total_seconds 内按 check_interval 分段睡眠，每段后检查 keep_running。
    当 keep_running() 返回 False（表示应停止）时立即返回；否则睡满 total_seconds 后返回。

    用于替代 time.sleep(interval)，使工作循环在收到停止信号后能在一个 check_interval 内退出。

    参数
    -----
    total_seconds : float
        计划等待的总时间（秒），对应业务上的间隔，如 config 中的 interval。
    keep_running : Callable[[], bool]
        无参可调用对象，返回 True 表示继续运行、应继续等待，返回 False 表示应停止、立即结束等待。
        例如: lambda: self._monitoring 或 lambda: self.monitor_started。
    check_interval : float, 默认 1.0
        每次睡眠的步长（秒），即检查 keep_running 的频率。不宜过大，否则停止响应仍会变慢。

    返回
    -----
    bool
        True：已睡满 total_seconds（未收到停止信号）。
        False：因 keep_running() 返回 False 而提前结束等待。

    示例
    -----
    >>> # 在监控循环中替代 time.sleep(interval)
    >>> while self._monitoring:
    ...     self._do_work()
    ...     if not interruptible_sleep(600, lambda: self._monitoring, check_interval=1.0):
    ...         break  # 收到停止，退出循环
    """
    if total_seconds <= 0:
        return True
    check_interval = max(0.01, min(check_interval, total_seconds))
    deadline = time.monotonic() + total_seconds
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return True
        chunk = min(check_interval, remaining)
        time.sleep(chunk)
        if not keep_running():
            return False

注意，使用Callable来传递self._started，类似于C中的引用方法，因为self._started是不断变化的，直接传递一个started参数，则工具内部，started一直为False，起不到检测self._started是否结束的作用。

import threading
import time

class Monitor:
    def __init__(self):
        self._started = False
        self._lock = threading.Lock()
        self._interval = 600
        self._thread = None

    def start(self, interval=None):
        with self._lock:
            if self._started:
                return
            self._started = True
            if interval is not None:
                self._interval = interval
            self._thread = threading.Thread(target=self._run, daemon=True)
            self._thread.start()

    def stop(self):
        with self._lock:
            self._started = False

    def is_running(self):
        with self._lock:
            return self._started

    def _run(self):
        while self.is_running():
            self.monitor()
            # 用 interruptible_sleep 替代 time.sleep
            if not interruptible_sleep(self._interval, self.is_running, check_interval=1.0):
                break

    def monitor(self):
        # 具体的监控逻辑在此处实现
        print("正在监控...")