如何通过schedule定时任务在非__main__作用域运行多进程/线程函数

你当前的调度逻辑跑不起来，主要是三个常见错误：

• schedule.do()调用目标函数时不会自动读取外部的tickers变量，你没有给foo传参，触发调度时会直接报参数缺失错误
• 最初版本在foo内部每次执行都新建ProcessPoolExecutor和asyncio事件循环，进程创建开销极大，还容易出现事件循环跨上下文冲突
• 调度主循环没有加休眠，会占满CPU空转

对应你调整后的executor.map版本，不需要混用asyncio，逻辑更稳定：

import time
import schedule
from concurrent.futures import ProcessPoolExecutor

# 全局初始化一次进程池，避免每次调度重复创建销毁进程
executor = ProcessPoolExecutor(max_workers=2)

def get_data(ticker):
    # 替换为你实际的行情拉取/处理逻辑
    print(f"开始处理标的: {ticker}")
    time.sleep(1.5) # 模拟IO耗时
    print(f"标的处理完成: {ticker}")

def foo(tickers):
    task_args = (ticker for ticker in tickers)
    # 提交批量任务到进程池，提交后立刻返回不阻塞调度循环
    executor.map(get_data, task_args)
    # 如果需要等待本轮所有任务执行完再退出foo，改成: list(executor.map(get_data, task_args))

if __name__ == "__main__":
    # 替换为你实际的标的列表
    tickers = ["AAPL", "MSFT", "TSLA", "NVDA"]
    # 关键：调度时把tickers作为位置参数传给do，触发时会自动传入foo
    schedule.every().minute.at(":00").do(foo, tickers)

    # 启动调度主循环
    while True:
        schedule.run_pending()
        time.sleep(1) # 加1秒休眠，避免CPU空转

• 测试阶段可以把调度规则改成schedule.every(10).seconds.do(foo, tickers)，不用等整分就能验证逻辑
• 如果你的get_data是IO密集型任务（比如拉接口、读文件），把ProcessPoolExecutor换成ThreadPoolExecutor性能更好，进程池更适合CPU密集型计算场景
• 如果单次foo执行时间可能超过1分钟，建议加一个全局运行标记，避免上一轮任务没跑完下一轮又重复提交，导致任务堆积：

  is_running = False
  def foo(tickers):
      global is_running
      if is_running:
          return
      is_running = True
      try:
          task_args = (ticker for ticker in tickers)
          list(executor.map(get_data, task_args))
      finally:
          is_running = False
  

• 不要在同步调度的foo函数里随意调用asyncio.get_event_loop()，schedule本身是同步调度框架，混用asyncio逻辑很容易出现跨线程/跨进程的事件循环错误。

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