Запуск функции каждые 60 сек в Python: альтернатива cronjob
Быстрый ответ
Для периодического запуска функций в Python можно использовать threading.Timer
. Этот класс позволяет выполнить функцию через заданный промежуток времени и может быть рекурсивно вызван. Рассмотрим пример:
from threading import Timer
def repeater(interval, function):
Timer(interval, repeater, [interval, function]).start()
function()
# Пример использования: функция `my_task` будет вызываться каждые 2 секунды.
def my_task():
print("Задача выполнена")
repeater(2, my_task)
В данном примере функция my_task
будет вызываться каждые две секунды посредством рекурсии в функции repeater
.
Продвинутые методы планирования: Выбор подходящего инструмента
Задача периодического запуска функций часто встречается, и хотя применение threading.Timer
достаточно удобно, в Python есть ряд более мощных инструментов для более сложных случаев.
Планирование событий с использованием sched
import sched, time
scheduler = sched.scheduler(time.monotonic, time.sleep)
def scheduled_job():
print('Выполнена запланированная функция.')
def scheduler_tool(exec_interval, func, scheduled_event=None):
if scheduled_event:
scheduler.cancel(scheduled_event)
new_event = scheduler.enter(exec_interval, 1, func)
scheduler.run(blocking=False)
return new_event
current_event = scheduler_tool(5, scheduled_job)
Учет монотонного времени увеличивает точность планирования, так как минимизируется влияние изменения системного времени.
AsyncIO: вариант для многозадачности
Для I/O-агрессивных или сетевых задач asyncio — это встроенная библиотека, которая позволяет писать асинхронный код с помощью ключевых слов async и await.
import asyncio
async def periodic_operation(interval):
while True:
await asyncio.sleep(interval)
print("Выполняется периодическая операция")
loop = asyncio.get_event_loop()
loop.run_until_complete(periodic_operation(3))
Визуализация
Механизм запуска функции каждые x секунд выглядит подобно работе часов:
def regular_operation():
tick() # Начало функции
time.sleep(x) # Пауза на x секунд
tock() # Снова начало функции
Это непрерывное движение:
Циферблат: 🕒🕒🕒🕒🕒
Сердцебиение: ⏱️ → tick → ⏳ → tock → ⏱️
Интервалы: Каждые x секунд...
Каждый "tick-tock" образует цикл, а задача выполняется регулярно с постоянным интервалом.
Избегание типичных ошибок
Имеются нюансы при реализации периодических запусков функций, которые необходимо учесть для обеспечения надежности и эффективности кода.
Забудьте про циклы while с time.sleep()
Самый простой способ — использовать цикл while
с time.sleep(interval)
, однако этот подход может привести к смещению рассчетного времени вызова функции из-за продолжительности её выполнения.
Запуск функций в потоках для сохранения интервалов
Если выполнение функции занимает много времени и может превышать заданный интервал, стоит использовать потоки, чтобы сохранить консистентность интервалов:
import threading
def run_at_intervals(interval, function):
next_call = time.time()
while True:
threading.Thread(target=function).start()
next_call += interval
time.sleep(max(0, next_call – time.time()))
def long_running_function():
print("Функция 'long_running_function' стартовала")
run_at_intervals(10, long_running_function)
Twisted для высокой точности и устойчивости
Если требуется высокая точность планирования и надежность, стоит обратить внимание на Twisted's LoopingCall
.
from twisted.internet import task, reactor
def periodic_execution():
print("Twisted: активируется периодическое выполнение функций.")
interval = 6.0
lc = task.LoopingCall(periodic_execution)
lc.start(interval)
reactor.run()
Гибкое управление «на ходу»
Классы по типу RepeatedTimer
позволяют корректировать планирование в реальном времени и предотвращать перекрытие вызовов функций.
Простота и функциональность
Нецелесообразно пренебрегать простыми решениями исключительно на Python без дополнительных зависимостей. Однако, если речь идет о сложных сценариях, то стоит обратить внимание на такие библиотеки как APScheduler
или Celery
для распределенного выполнения задач.
Полезные материалы
- time — доступ к времени и его преобразования в Python – документация по модулю time.
- threading — многопоточность в Python – официальное руководство по threading.
- Техники работы с таймерами – подробная статья по работе с таймером в Python.
- sched — планировщик событий в Python – руководство по работе с модулем sched.
- Полное руководство по async IO в Python – подробный туториал по асинхронному коду с помощью
asyncio
. - Celery — для выполнения периодических задач – документация по Celery.
- Применение time.sleep() в Python – статья о использовании time.sleep для создания задержек.