Python实现定时任务的几种方法

import time

def func():
    print("func executed")

def timer():
    while True:
        func()
        time.sleep(5)

timer()

# 每隔5s执行一次func函数

import datetime
date = datetime.datetime.now().strftime("%Y-%m-%d %H:%m")
if date == '2021-12-19 17:12':
    func()

import datetime
from threading import Timer

def func(para):
    print(para)
    print("func exexuted")

def timer(delay):
    hit = "timer called func"
    t = Timer(delay, func, (hit,))  # 三个参数分别是：延迟时间 调用函数 (传入调用函数的参数（必须是tuple）)
    t.start()

timer(3.0)

import sched
scheduler = sched.scheduler(time.time, time.sleep) # 两个参数分别为 返回时间戳的函数，默认就是time.time 和 定时未到达之前执行何种函数阻塞，默认就是time.sleep

def func(para):
    print(para)
    print("func executed")

def timer(delay):
    hit = "scheduled func"
    scheduler.enter(delay, 0, func, (hit,)) # 四个参数分别为 延迟时间 优先级(如果调度包含多个任务，需指定优先级) 调用函数 传入参数
    scheduler.run()

timer(3)

pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler  # 阻塞方法

def job(para):
	print(parm)
    print("job executed")

scheduler = BlockingScheduler()
scheduler.add_job(job, ['定时器任务'], 'interval', seconds=5)  # 第一个参数指定要执行的函数 第二个参数指定传入的参数 第二个参数指定调度方式 也可以是'date'或者'cron' 第三个参数是'interval'间隔调度模式下间隔的时间 也可以指定hours或者minutes
scheduler.start()

from apscheduler.schedulers.background import BackgroundScheduler  # 非阻塞方法

def job(para):
	print(parm)
    print("job executed")

scheduler = BackgroundScheduler()
scheduler.add_job(job, ['定时器任务'], 'interval', seconds=5)
scheduler.start()

from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 参数
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 输出时间
def job():
    print("jon executed")

# 存储方式配置
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
# 执行器配置
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
# 直接在被调度函数前加装饰器
@scheduler.scheduled_job('interval', id='job_id', seconds=5)
def job_function():
    print("Hello World")

scheduler.start()

from apscheduler.schedulers.background import BackgroundScheduler  # 非阻塞方法

def job(para):
	print(parm)
    print("job executed")

scheduler = BackgroundScheduler()
scheduler.add_job(job, ['定时器任务'], 'interval', hours=5, timezone='asia/shanghai')
scheduler.start()

from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler

def job(para):
    print(para)
    
scheduler = BlockingScheduler()
# The job will be executed on December 19th, 2021
scheduler.add_job(job, 'date', run_date=date(2021, 12, 19), args=['定时器任务'])
scheduler.add_job(job, 'date', run_date=datetime(2021, 12, 19, 18, 39, 5), args=['定时器任务']) # 年月日 时分秒
scheduler.add_job(job, 'date', run_date='2021-12-19 18:39:05', args=['定时器任务']) # 上面的另一种形式

# The 'date' trigger and datetime.now() as run_date are implicit
# 上面的英文的意思是date触发器是默认的触发器，此时此刻是默认指定的运行时间 所以可以省略 指定的任务将会立即执行
scheduler.add_job(job, args=['text'])
scheduler.start()

crontab t1 t2 t3 t4 t5 job

*    *    *    *    *
-    -    -    -    -
|    |    |    |    |
|    |    |    |    +----- 星期中星期几 (0 - 6) (星期天 为0)
|    |    |    +---------- 月份 (1 - 12) 
|    |    +--------------- 一个月中的第几天 (1 - 31)
|    +-------------------- 小时 (0 - 23)
+------------------------- 分钟 (0 - 59)

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print("Hello World")
    
# BlockingScheduler
scheduler = BlockingScheduler()
# Schedules job_function to be run on the third Friday
# of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
scheduler.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# Runs from Monday to Friday at 5:30 (am) until 2021-12-30 00:00:00
scheduler.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2021-12-30')
scheduler.start()

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
# use id
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

# 暂停
apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()
# 恢复
apscheduler.job.Job.resume()
apscheduler.schedulers.base.BaseScheduler.resume_job()

job.modify(max_instances=6, name='Alternated name')
modify_job('my_job_id', trigger='cron', minute='*/5')

scheduler.shutdown()
scheduler.shutdown(wait=False)

def my_listener(event):
    if event.exception:
        print('The job crashed :(')
    else:
        print('The job worked :)')
# 添加监听器
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)