You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python函数调度问题:如何确保同一时间仅运行一个函数

确保Python函数串行执行并延迟触发后续任务的解决方案

需求回顾

需要实现以下逻辑:

  • 对参数列表[1,2,3,4,5],依次串行执行A(1)→A(2)→A(3)→A(4)→A(5),每个A(arg)执行完成后立即启动下一个A
  • 每个A(arg)完成后启动10秒定时器,到期后执行B(arg)
  • 核心约束:同一时间只能有一个函数(A或B)在执行,若B的延迟到期时已有函数运行,需等待该函数执行完毕再启动B

原代码使用threading.Timer直接启动B,导致B在新线程中独立运行,与主线程的A执行重叠,违反了串行约束。

修正后的代码

import threading
import time

def A(arg):
    print(f"A({arg}) started at {time.time():.2f}")
    time.sleep(4)
    print(f"A({arg}) finished at {time.time():.2f}")

def B(arg):
    print(f"B({arg}) started at {time.time():.2f}")
    time.sleep(1)
    print(f"B({arg}) finished at {time.time():.2f}")

# 全局锁,确保同一时间仅一个函数执行
execution_lock = threading.Lock()

def delayed_execute_B(arg):
    # 获取锁后再执行B,保证串行执行
    with execution_lock:
        B(arg)

def run_a_then_schedule_b(arg):
    # 执行A时先获取锁,保证A的串行执行
    with execution_lock:
        A(arg)
    # A执行完成后,启动10秒定时器,到期后触发带锁的B执行
    timer = threading.Timer(10, delayed_execute_B, args=(arg,))
    timer.start()

args = [1, 2, 3, 4, 5]

for arg in args:
    run_a_then_schedule_b(arg)

关键逻辑说明

  1. 全局锁控制串行:通过threading.Lock()创建全局锁,所有函数(A和B)必须获取锁后才能执行,从根本上避免多函数并行运行。
  2. A的串行执行:主线程循环调用run_a_then_schedule_b时,前一个A执行完毕释放锁后,下一个A才能获取锁启动,保证A任务按顺序串行执行。
  3. 延迟B的串行触发:定时器到期后,delayed_execute_B会先尝试获取锁,若此时有A或其他B在运行,会自动等待锁释放后再执行B,满足延迟且不重叠的要求。

内容的提问来源于stack exchange,提问作者MauroL

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.11 02:03:38