Python中结合Scheduling与多进程实现并行执行的问题求助

问题描述 #

需要编写Python脚本，让函数run_script既按指定时间调度执行，又能以不同参数通过多进程并行运行。尝试过线程和多进程，但调度正常的情况下，函数始终串行执行，无法同时输出不同参数的结果。

预期输出 #

This is the 1 script...
This is the 2 script...
This is the 3 script...
This is the 4 script...
This is the 1 script...
This is the 2 script...
This is the 3 script...
This is the 4 script...
...

实际输出 #

This is the 1 script...
This is the 1 script...
This is the 1 script...
...
This is the 2 script...
This is the 2 script...
This is the 2 script...
...
This is the 3 script...
This is the 3 script...
This is the 3 script...
...
This is the 4 script...
This is the 4 script...
This is the 4 script...
...

原代码 #

import schedule, time
from multiprocessing import Process


def run_script(n):
    p = 1
    while p < 1000:
        print(f"This is the {n} script...")
        p += 1

def procc(n):
    proc = Process(target=run_script, args=(n,))
    proc.start()
    proc.join()

rows = [1,2,3,4]

print("Assigning Schedules")
for row in rows:        
    schedule.every(10).seconds.do(procc, row)

print("Starting schedule...")
while True:
    schedule.run_pending()
    time.sleep(1)

问题分析 #

核心问题出在procc函数中的proc.join()：

• schedule库的任务是在主进程串行执行的，每次只会运行一个到期的任务。
• proc.join()会阻塞当前的调度任务，直到该子进程完全执行完毕才会返回，导致下一个调度任务（比如procc(2)）必须等procc(1)的子进程跑完才能启动，最终所有子进程串行执行。

解决方案 #

方案1：移除阻塞的join调用 #

直接去掉proc.join()，让子进程启动后立刻返回，调度任务可以快速完成，多个子进程就能并行运行：

import schedule, time
from multiprocessing import Process


def run_script(n):
    p = 1
    while p < 1000:
        print(f"This is the {n} script...")
        p += 1

def procc(n):
    proc = Process(target=run_script, args=(n,))
    proc.start()
    # 移除proc.join()，避免阻塞调度任务的执行流程

rows = [1,2,3,4]

print("Assigning Schedules")
for row in rows:        
    schedule.every(10).seconds.do(procc, row)

print("Starting schedule...")
while True:
    schedule.run_pending()
    time.sleep(1)

方案2：批量启动多进程（更贴合预期输出） #

如果希望每次调度时同时启动所有参数的进程，可以编写一个批量启动的函数，让调度器触发这个函数，而非单独调度每个参数：

import schedule, time
from multiprocessing import Process


def run_script(n):
    p = 1
    while p < 1000:
        print(f"This is the {n} script...")
        p += 1

def run_all_scripts():
    procs = []
    # 批量创建并启动所有子进程
    for n in [1,2,3,4]:
        proc = Process(target=run_script, args=(n,))
        procs.append(proc)
        proc.start()
    # 若需要等待本次所有进程完成后再进行下一次调度，可取消下方注释
    # for proc in procs:
    #     proc.join()

print("Assigning Schedules")
# 每10秒触发一次批量启动任务
schedule.every(10).seconds.do(run_all_scripts)

print("Starting schedule...")
while True:
    schedule.run_pending()
    time.sleep(1)

补充说明 #

• 若使用线程（threading.Thread）遇到同样问题，原因类似：thread.join()会阻塞调度任务，移除后即可并行。
• 若run_script逻辑复杂、资源占用高，建议使用多进程而非线程，避免GIL限制。

内容的提问来源于stack exchange，提问作者Roelof Coertze