如何实现多频次bash脚本在一分钟内的均匀间隔调度

我现在有一个tasks.yaml配置文件，里面定义了多个bash脚本，每个脚本需要在一分钟内运行指定的次数：

every_minute:
    - name: test0
      path: test0.sh
      times_per_min: 3
    - name: test1
      path: test1.sh
      times_per_min: 2

原本我用Python写了一段调度代码，思路是把每个脚本按重复次数塞进任务列表，然后按固定间隔依次执行：

import subprocess
import time
import yaml

TASKS = []
PROCESSES = []  # 补充原代码遗漏的变量定义

with open("tasks.yaml") as file:
  measurements = yaml.safe_load(file)

for every_min in measurements["every_minute"]:
  for _ in range(every_min["times_per_min"]):
    TASKS.append(every_min["path"])

interval = 60 / len(TASKS)

for task in TASKS:
  try:
    PROCESSES.append(subprocess.Popen([task]))
  except Exception as e:
    print(f"An error occurred: {e}")  # 优化错误提示，展示具体异常
  time.sleep(interval)

但这样生成的任务列表是连续重复同一个脚本：

print(TASKS)
# 输出: ['test0.sh', 'test0.sh', 'test0.sh', 'test1.sh', 'test1.sh']

我想要的是均匀穿插调度的效果，让不同脚本尽量交替执行，最终任务列表应该是：

['test0.sh', 'test1.sh', 'test0.sh', 'test1.sh', 'test0.sh']

而且这个逻辑要能适配任意数量的任务、任意的times_per_min值，不能只针对当前示例。

解决方案：按“轮询优先级”生成均匀穿插的任务列表 #

核心思路是：不要先把同一个脚本的所有实例都塞进列表，而是每次从所有任务中挑选还没达到执行次数的脚本，按轮询的方式依次添加，直到所有任务的执行次数都满足要求。

具体实现代码如下：

import subprocess
import time
import yaml

PROCESSES = []

with open("tasks.yaml") as file:
  measurements = yaml.safe_load(file)

# 先整理任务元数据：保存每个脚本的路径和剩余需要执行的次数
task_queue = [
    {"path": item["path"], "remaining": item["times_per_min"]}
    for item in measurements["every_minute"]
]

TASKS = []

# 循环生成均匀穿插的任务列表
while task_queue:
    # 遍历当前还有剩余执行次数的任务（用list(range)避免遍历时队列长度变化的问题）
    for idx in list(range(len(task_queue))):
        task = task_queue[idx]
        TASKS.append(task["path"])
        task["remaining"] -= 1
        # 若该任务已完成所有执行次数，从队列中移除
        if task["remaining"] == 0:
            del task_queue[idx]

# 计算执行间隔，避免除以0的情况
total_tasks = len(TASKS)
interval = 60 / total_tasks if total_tasks > 0 else 0

# 执行任务
for task_idx, task in enumerate(TASKS):
    try:
        PROCESSES.append(subprocess.Popen([task]))
    except Exception as e:
        print(f"执行任务 {task} 时出错: {e}")
    # 最后一个任务不需要sleep，避免额外等待间隔
    if task_idx != total_tasks - 1 and total_tasks > 0:
        time.sleep(interval)

逻辑验证 #

针对示例中的配置，生成的TASKS列表正好符合预期：

print(TASKS)
# 输出: ['test0.sh', 'test1.sh', 'test0.sh', 'test1.sh', 'test0.sh']

如果测试更复杂的场景，比如：

every_minute:
    - name: test0
      path: test0.sh
      times_per_min: 4
    - name: test1
      path: test1.sh
      times_per_min: 3
    - name: test2
      path: test2.sh
      times_per_min: 2

生成的任务列表会是：

['test0.sh', 'test1.sh', 'test2.sh', 'test0.sh', 'test1.sh', 'test2.sh', 'test0.sh', 'test1.sh', 'test0.sh']

完全符合均匀穿插的预期。

代码解释 #

1. 任务队列初始化：用字典保存每个任务的路径和剩余执行次数，方便跟踪执行进度。
2. 轮询生成列表：循环遍历剩余任务，每添加一个任务就减少其剩余次数，完成次数的任务会被从队列中移除，直到所有任务都达标。
3. 执行逻辑优化：最后一个任务执行后不再sleep，避免多等待一个间隔；同时优化了错误提示，能看到具体的异常信息。

这样的实现完全通用，不管有多少个任务、每个任务需要执行多少次，都能保证任务被尽可能均匀地穿插调度，满足一分钟内的分布要求。

备注：内容来源于stack exchange，提问作者Jesper.Lindberg