• python内置定时任务
  • while True + sleep
  • Timeloop
  • Timer
  • sched

python内置定时任务

while True + sleep

import datetime


def task_run(*args, **kwargs):
    print(f"耗时操作....{datetime.datetime.now().strftime('%Y%m%d_%H:%M:%S')}")
    
def cron_and_sleep():
    '''
    while True + sleep
    :return:
    '''
    while True:
        task_run()  # 执行的任务
        time.sleep(3)

Timeloop

import datetime
from timeloop import Timeloop

def task_run(*args, **kwargs):
    print(f"耗时操作....{datetime.datetime.now().strftime('%Y%m%d_%H:%M:%S')}")

def cron_and_timeloop():
    '''
    Timeloop
    :return:
    '''
    t = Timeloop()
    t._add_job(task_run, datetime.timedelta(seconds=3), (1,))
    t.start(block=True)

Timer

import datetime
from threading import Timer


def task_run(*args, **kwargs):
    print(f"耗时操作....{datetime.datetime.now().strftime('%Y%m%d_%H:%M:%S')}")


class CronTimer(Timer):
    def run(self):
        while not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
            self.finished.wait(self.interval)
            

def cron_and_timer():
    '''
    timer
    threading模块提供了一个定时器触发的函数Timer是一个非阻塞函数  任务执行一次 需要使用循环保证在特定时间间隔调用
    :return:
    '''
    t = CronTimer(3, task_run, (1,))
    t.start()
    # t.cancel()  # 停止还在运行的任务

sched

import sched
import time
import datetime

def cron_and_sched():
    s = sched.scheduler(time.time, time.sleep)
    
    def task(*args, **kwargs):
        """任务"""
        try:
            print(f"耗时操作....{datetime.datetime.now().strftime('%Y%m%d_%H:%M:%S')}")
        finally:
            s.enter(3, 1, task, (1, ))

    s.enter(3, 1, task, (1,))
    s.run()