Параллельный запуск функции по расписанию с разными параметрами
Как на python >3.8 реализовать параллельный запуск функций по расписанию, но с разными параметрами и чтобы работа скрипта не была привязана к версии операционой системы?
Ниже приведен пример конфиг файла. Скрип на python будет считывать конфиг, а далее в зависимости от заданной в конфиге периодичности должен запускать функцию fnc с заданным параметром в параллельном режиме. Для примера если пользователь задал в конфиге задал параметры 1m, 1h, 1d то в 00:00:00 каждого дня должно параллельно выполниться функции с параметрами fnc(timeframe = 1m), fnc(timeframe = 1h), fnc(timeframe = 1d). Как реализовать заданный функционал и максимально оптимально и правильно?
ini file
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
Ответы (2 шт):
Можно, к примеру, опереться на монотонный таймер 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 - это асинхронный цикл, который в этом случае получает сигналы монотонного таймера от ядра системы.
dispatcher - в первую очередь запланирует себя к выполнению в 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()
Вариант через 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()