主页 / 问题 / 1226413
Accepted
Yuriy Tigiev
Asked:2022-01-02 19:27:22 +0000 UTC2022-01-02 19:27:22 +0000 UTC

并行启动具有不同参数的预定函数

  • 772

如何在 python > 3.8 中实现计划函数的并行启动，但使用不同的参数，以便脚本不绑定到操作系统的版本？

下面是一个示例配置文件。python 脚本将读取配置，然后根据配置中指定的频率，它应该以并行模式运行具有指定参数的 fnc 函数。例如，如果用户在 config 中指定了参数 1m, 1h, 1d，那么在每天的 00:00:00，具有参数 fnc(timeframe = 1m), fnc(timeframe = 1h), fnc( timeframe = 1d) 应该并行执行。如何以最优化和正确的方式实现给定的功能？

.ini 文件

1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
python

2 个回答

  1. Best Answer

    例如，您可以依靠单调的计时器asyncio进行concurrent.futures.Executor并行化。

    import asyncio
import datetime
import time
import re
import concurrent.futures

tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(loop, pool, crontab):
    time_seconds = time.time()
    delay_seconds = (time_seconds//60+1) * 60 - time_seconds
    loop.call_later(delay_seconds, dispatcher, loop, pool, crontab)
    minute = time_seconds // 60 + TIMEZONE
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            pool.submit(callback, **{name: value})


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(line) for line in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab


pool = concurrent.futures.ThreadPoolExecutor(max_workers=5)
loop = asyncio.get_event_loop()
dispatcher(loop, pool, get_crontab())
loop.run_forever()

    在我的电脑上，时钟的延迟结果在 0.003-0.060 秒的范围内

    get_crontab - 将示例文本中的格式转换为列表：间隔、函数、参数、值。

    pool 是 5 个线程，它们几乎同时运行这些函数。ProcessPoolExecutor如果要跨所有内核卸载，请替换为。

    loop 是一个异步循环，在这种情况下，它从系统内核接收单调定时器信号。

    调度程序 - 首先安排自己在下一分钟的 00 秒执行。然后它将执行列表中应在当前分钟内执行的功能。

    pool.submit - 在池线程之一中并行运行该函数。

    竞争选项（不是并行的，而是异步的）

    import asyncio
import datetime
import time
import re
import concurrent.futures
import inspect
import functools

tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


async def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(loop, crontab):
    time_seconds = time.time()
    delay_seconds = (time_seconds//60+1) * 60 - time_seconds
    loop.call_later(delay_seconds, dispatcher, loop, crontab)
    minute = time_seconds // 60 + TIMEZONE
    if time_seconds % 60 > 5:
        return 
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            if inspect.iscoroutinefunction(callback):
                loop.create_task(callback(**{name: value}))
            elif inspect.isfunction(callback):
                loop.call_soon(functools.partial(callback, **{name: value}))


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(m) for m in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab


async def main():
    dispatcher(loop, get_crontab())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()
    • 5

  2. 选项通过time.sleep和multiprocessing

    import multiprocessing
import datetime
import time
import re


tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(pool, crontab, time_seconds):
    minute = time_seconds // 60 + TIMEZONE
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            pool.apply_async(callback, kwds={name: value})


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(m) for m in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab

pool = multiprocessing.Pool()
crontab =  get_crontab()


while True:
    time_seconds = time.time()
    dispatcher(pool, crontab, time_seconds)
    delay_seconds = (time_seconds//60+1) * 60 - time.time()
    time.sleep(delay_seconds)


pool.terminate()
pool.join()
    • 4

相关问题