Запуск функции каждые 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 для распределенного выполнения задач.

Полезные материалы

1. time — доступ к времени и его преобразования в Python – документация по модулю time.
2. threading — многопоточность в Python – официальное руководство по threading.
3. Техники работы с таймерами – подробная статья по работе с таймером в Python.
4. sched — планировщик событий в Python – руководство по работе с модулем sched.
5. Полное руководство по async IO в Python – подробный туториал по асинхронному коду с помощью asyncio.
6. Celery — для выполнения периодических задач – документация по Celery.
7. Применение time.sleep() в Python – статья о использовании time.sleep для создания задержек.