多站点场景下如何无漂移地每隔X秒调度迭代函数？

问题根源分析 #

1. 原代码中，主线程的while True循环每次都会创建新线程池并提交所有站点的任务，若logs函数内部包含while 1循环，会导致同一个站点被多个线程同时处理，进而产生重复日志。
2. 使用time.sleep(x)的相对休眠方式，会因为任务处理耗时（文件检查、读写、删除操作）不断累积，导致每次执行的时间持续向后漂移；站点数量越多，单轮任务总耗时越长，漂移现象越严重。

方案1：全局统一绝对时间调度（推荐） #

将调度逻辑放在主线程，通过固定的目标时间点触发任务，彻底避免漂移；同时复用线程池，减少资源开销。

import time
import os
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

def client_list():
    sites = pd.read_csv('sites')
    return sites['Site'].tolist()

def process_site(site):
    """处理单个站点的状态检查与日志记录"""
    hit_path = os.path.join(site, 'target', 'hit')
    stamp = time.strftime('%Y-%m-%d,%H:%M:%S')
    log_path = os.path.join(site, 'log')
    
    # 用with语句自动管理文件句柄，避免手动close的潜在问题
    with open(log_path, 'a') as log_file:
        if os.path.isfile(hit_path):
            log_file.write(f",{stamp},{site},hit\n")
            try:
                os.remove(hit_path)
            except OSError:
                # 处理文件已被其他线程/进程删除的异常
                pass
        else:
            log_file.write(f",{stamp},{site},miss\n")

if __name__ == '__main__':
    interval = 5  # 设定执行间隔（秒）
    sites = client_list()
    
    # 复用线程池，避免每次循环重复创建销毁
    with ThreadPoolExecutor() as executor:
        next_run_time = time.time()
        while True:
            # 等待到预设的目标执行时间
            time.sleep(max(0, next_run_time - time.time()))
            # 批量提交所有站点的处理任务
            executor.map(process_site, sites)
            # 更新下一次执行的目标时间（基于上一次目标时间+间隔，而非当前时间）
            next_run_time += interval

核心优势：

• 主线程控制全局调度时间，确保每次任务都对齐预设时间点，彻底消除漂移。
• 每个站点仅被单次任务处理，不会出现重复日志。
• 优化文件操作逻辑，避免资源泄漏和异常报错。

方案2：单站点独立绝对时间调度 #

若需要为不同站点设置独立的调度周期，可让每个站点线程自行维护目标时间：

import time
import os
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

def client_list():
    sites = pd.read_csv('sites')
    return sites['Site'].tolist()

def run_site_scheduler(site, interval=5):
    """单个站点的独立定时调度器"""
    next_run_time = time.time()
    while True:
        time.sleep(max(0, next_run_time - time.time()))
        
        # 执行站点状态检查与日志记录
        hit_path = os.path.join(site, 'target', 'hit')
        stamp = time.strftime('%Y-%m-%d,%H:%M:%S')
        log_path = os.path.join(site, 'log')
        
        with open(log_path, 'a') as log_file:
            if os.path.isfile(hit_path):
                log_file.write(f",{stamp},{site},hit\n")
                try:
                    os.remove(hit_path)
                except OSError:
                    pass
            else:
                log_file.write(f",{stamp},{site},miss\n")
        
        # 更新下一次执行的目标时间
        next_run_time += interval

if __name__ == '__main__':
    sites = client_list()
    with ThreadPoolExecutor() as executor:
        # 为每个站点启动独立的调度线程
        for site in sites:
            executor.submit(run_site_scheduler, site)
        # 主线程保持存活，避免进程退出
        while True:
            time.sleep(3600)

适用场景：不同站点需要不同执行间隔的需求，每个站点仅由一个线程处理，无重复日志和漂移问题。

你的尝试失败原因 #

1. logs函数内部的while 1循环，加上主线程while True重复提交任务，导致同一个站点被多个线程并发处理，产生重复日志。
2. 漂移计算逻辑有误：num_calls初始值设为1，第一次休眠后就递增，导致time_period * num_calls的计算偏差，无法准确修正漂移。

内容的提问来源于stack exchange，提问作者Mark Goodwin