如何让函数部分代码暂停指定时长且不阻塞其他调度任务？

首先要明确：schedule模块是单线程运行的，所以如果在调度的函数里直接用time.sleep()，会阻塞整个调度循环，导致所有其他任务都被延迟。你之前尝试的线程方式没生效，大概率是因为线程的用法不对——应该在defN内部启动子线程来处理延迟逻辑，而不是在其他函数里调用它。

下面给你两种可行的解决方案，按推荐程度排序：

方案一：拆分逻辑为数据收集+缓存+定时处理（最推荐） #

这个方案逻辑清晰，不需要额外线程管理，还能保证数据线程安全。核心思路是：

• 用一个线程安全的缓存存储上一次收集的数据
• 每小时收集一次最新数据并更新缓存
• 每小时检查缓存中是否有历史数据，如果有就进行新旧数据运算

代码示例 #

import schedule
import time
import threading
import requests
from typing import Optional

# 线程安全的缓存，用于存储上一次收集的数据
last_collected_data: Optional[dict] = None
data_lock = threading.Lock()

def def1():
    url = "你的目标URL"
    response = requests.get(url)
    # 根据实际需求处理原始响应，这里假设返回JSON
    return response.json()

def def2():
    # 调用def1并执行相关处理，返回结构化数据（含时间戳方便追踪）
    raw_data = def1()
    processed_data = {
        "timestamp": time.time(),
        "content": raw_data
    }
    return processed_data

def collect_and_process():
    """每小时执行：收集新数据，与历史数据运算（第一次无历史数据则跳过）"""
    global last_collected_data
    
    # 收集当前最新数据
    new_data = def2()
    
    # 线程安全地读取并更新缓存
    with data_lock:
        old_data = last_collected_data
        last_collected_data = new_data
    
    # 第一次运行时没有历史数据，直接跳过运算
    if old_data is not None:
        # 这里写入你的新旧数据运算逻辑
        print(f"对比 {time.ctime(old_data['timestamp'])} 和 {time.ctime(new_data['timestamp'])} 的数据")
        # example: calculate_diff(old_data['content'], new_data['content'])
    else:
        print("首次运行，无历史数据可处理")

# 调度你的其他常规函数
schedule.every.hour.do(def1)
schedule.every.hour.do(def2)
# 调度defN对应的逻辑
schedule.every.hour.do(collect_and_process)

# 启动schedule主循环
while True:
    schedule.run_pending()
    time.sleep(1)

额外优化：如果网络请求耗时较长 #

如果def1中的requests.get()耗时比较久，会阻塞整个调度循环。这时候可以把def1的执行放到子线程里：

def run_def1_in_thread():
    threading.Thread(target=def1).start()

# 替换原来的调度
schedule.every.hour.do(run_def1_in_thread)

方案二：在defN内部用子线程处理延迟逻辑 #

如果你不想拆分函数，可以直接在defN里启动子线程，让子线程负责延迟1小时后处理数据，这样defN本身会立即执行完毕，不会阻塞调度循环。

代码示例 #

import schedule
import time
import threading
import requests

def def1():
    url = "你的目标URL"
    return requests.get(url).json()

def def2():
    raw_data = def1()
    return {"timestamp": time.time(), "content": raw_data}

def defN():
    # 先收集当前数据作为"旧数据"
    old_data = def2()
    
    # 定义延迟处理的逻辑
    def delayed_process():
        # 延迟1小时
        time.sleep(3600)
        # 收集1小时后的新数据
        new_data = def2()
        # 执行新旧数据运算
        print(f"延迟处理：对比 {time.ctime(old_data['timestamp'])} 和 {time.ctime(new_data['timestamp'])} 的数据")
        # example: calculate_diff(old_data['content'], new_data['content'])
    
    # 启动子线程执行延迟逻辑，不阻塞主线程
    threading.Thread(target=delayed_process, daemon=True).start()

# 调度所有函数
schedule.every.hour.do(def1)
schedule.every.hour.do(def2)
schedule.every.hour.do(defN)

# 启动调度循环
while True:
    schedule.run_pending()
    time.sleep(1)

为什么你之前的线程方式没生效？ #

你提到"在除defN外的其他函数中添加代码t = threading.Thread(target=defN); t.start()"，这种方式会导致defN被额外触发（比如def2执行时会启动一个defN线程，加上schedule调度的defN，相当于每小时执行多次），而且逻辑混乱。正确的做法是让schedule只调度defN一次，然后在defN内部用线程处理延迟逻辑。

内容的提问来源于stack exchange，提问作者User 147