如何在Python中实现单调计时器？兼容多版本与Linux平台

核心思路 #

要解决系统时间跳变导致休眠失效的问题，我们直接基于Linux系统调用clock_nanosleep实现单调时钟的非忙等休眠，再封装成与threading.Timer接口一致的类，确保兼容Python 2.7、3.6、3.8。

步骤1：实现跨版本单调sleep函数 #

通过ctypes直接调用GLibC的clock_nanosleep，该函数支持指定CLOCK_MONOTONIC时钟，完全不受系统时间跳变影响。

import ctypes
import ctypes.util

# 加载libc库
libc = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)

# Linux下CLOCK_MONOTONIC的常量值
CLOCK_MONOTONIC = 1

# 定义clock_nanosleep所需的timespec结构体
class Timespec(ctypes.Structure):
    _fields_ = [
        ('tv_sec', ctypes.c_long),
        ('tv_nsec', ctypes.c_long)
    ]

def monotonic_sleep(seconds):
    """基于CLOCK_MONOTONIC的非忙等休眠，不受系统时间修改影响"""
    sec = int(seconds)
    nsec = int((seconds - sec) * 10**9)
    
    req = Timespec(sec, nsec)
    rem = Timespec(0, 0)
    
    while True:
        # 调用clock_nanosleep，0表示使用相对时间
        res = libc.clock_nanosleep(CLOCK_MONOTONIC, 0, ctypes.byref(req), ctypes.byref(rem))
        if res == 0:
            break
        elif res == 4:  # EINTR，休眠被信号中断，用剩余时间重试
            req = rem
        else:
            # 抛出系统错误
            errno = ctypes.get_errno()
            raise OSError(errno, ctypes.strerror(errno))

步骤2：封装MonotonicTimer类 #

模仿threading.Timer的接口，继承threading.Thread，使用上述monotonic_sleep实现休眠，同时支持取消功能：

import threading

class MonotonicTimer(threading.Thread):
    """兼容Python 2.7/3.6/3.8的Linux单调计时器，接口与threading.Timer一致"""
    def __init__(self, interval, function, args=None, kwargs=None):
        super(MonotonicTimer, self).__init__()
        self.interval = interval
        self.function = function
        self.args = args or []
        self.kwargs = kwargs or {}
        self.finished = threading.Event()

    def cancel(self):
        """取消计时器"""
        self.finished.set()

    def run(self):
        # 先检查是否已取消，未取消则执行单调休眠
        if not self.finished.is_set():
            try:
                monotonic_sleep(self.interval)
            except:
                return
            # 休眠完成后再次检查是否被取消，未取消则执行目标函数
            if not self.finished.is_set():
                self.function(*self.args, **self.kwargs)

关键说明 #

1. 单调时钟可靠性：clock_nanosleep(CLOCK_MONOTONIC)使用系统启动后的累计运行时间，无论系统时间向前/向后跳变，都不会影响休眠时长。
2. 非忙等实现：该系统调用会将线程挂起，直到指定的单调时间到达，不会占用CPU资源。
3. 版本兼容性：ctypes在Python 2.7及所有3.x版本中都可用，直接调用系统调用避开了Python标准库中sleep/wait的版本差异问题。
4. 取消机制：通过threading.Event实现取消，与原生Timer的行为一致，确保可以安全终止未触发的计时器。

使用示例 #

def greet(name):
    print(f"Hello {name}!")

# 创建一个3.5秒后执行的计时器
timer = MonotonicTimer(3.5, greet, args=("Bob",))
timer.start()

# 如需取消，直接调用cancel方法
# timer.cancel()

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