文章首页 Web开发 Windows开发 编程语言 数据库技术 移动平台 系统服务 微信 设计 布布扣 其他

python--实现定时任务

时间：2020-07-22 14:37:16   收藏：0   阅读：72

Python 实现定时任务

参考链接

一、循环 sleep

这种方式最简单，在循环里面放入要执行的任务，然后 sleep 一段时间再执行

from datetime import datetime
import time
# 每n秒执行一次
def timer(n):
    while True:
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        time.sleep(n)
# 5s
timer(5)

这个方法的缺点是：只能执行固定间隔时间的任务，如果有定时任务就无法完成，比如早上六点半喊我起床。并且 sleep 是一个阻塞函数，也就是说 sleep 这一段时间，啥都不能做。对服务器性能的损耗。

二、threading模块中的Timer

Timer 函数第一个参数是时间间隔（单位是秒，只有秒），第二个参数是要调用的函数名，第三个参数是调用函数的参数(tuple)。

from datetime import datetime
from threading import Timer
# 打印时间函数
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    t = Timer(inc, printTime, (inc,))
    t.start()
# 5s
printTime(5)

三、使用sched模块

sched 模块是 Python 内置的模块，它是一个调度（延时处理机制），每次想要定时执行某任务都必须写入一个调度。

import sched
import time
from datetime import datetime
# 初始化sched模块的 scheduler 类
# 第一个参数是一个可以返回时间戳的函数，第二个参数可以在定时未到达之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被周期性调度触发的函数
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    schedule.enter(inc, 0, printTime, (inc,))
# 默认参数60s
def main(inc=60):
    # enter四个参数分别为：间隔事件、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数，
    # 给该触发函数的参数（tuple形式）
    schedule.enter(0, 0, printTime, (inc,))
    schedule.run()
# 10s 输出一次
main(10)

sched 使用步骤如下：

四、定时任务框架APScheduler学习详解

参考链接

一、APScheduler简介

二、安装

使用 APScheduler 需要安装

pip install apscheduler

三、APScheduler有四种组成部分：

四、简单应用

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
def my_job():
    print time.strftime(‘%Y-%m-%d %H:%M:%S‘, time.localtime(time.time()))
 
sched = BlockingScheduler()
# 每隔5s执行一次my_job函数，输出当前时间信息
sched.add_job(my_job, ‘interval‘, seconds=5)
# 时间： 周一到周五每天早上6点半, 执行my_job
sched.add_job(my_job, ‘cron‘, day_of_week=‘1-5‘, hour=6, minute=30)
sched.start()

# 注释
代码中的 BlockingScheduler 是什么呢？

BlockingScheduler是APScheduler中的调度器，APScheduler 中有两种常用的调度器，BlockingScheduler 和 BackgroundScheduler，当调度器是应用中唯一要运行的任务时，使用 BlockingSchedule，如果希望调度器在后台执行，使用 BackgroundScheduler。

BlockingScheduler: use when the scheduler is the only thing running in your process
BackgroundScheduler: use when you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application
AsyncIOScheduler: use if your application uses the asyncio module
GeventScheduler: use if your application uses gevent
TornadoScheduler: use if you’re building a Tornado application
TwistedScheduler: use if you’re building a Twisted application
QtScheduler: use if you’re building a Qt application

五、任务操作

1.添加任务

上面是通过add_job()来添加任务，另外还有一种方式是通过scheduled_job()修饰器来修饰函数

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
sched = BlockingScheduler()
 
@sched.scheduled_job(‘interval‘, seconds=5)
def my_job():
    print time.strftime(‘%Y-%m-%d %H:%M:%S‘, time.localtime(time.time()))
 
sched.start()

上述代码创建了一个 BlockingScheduler，并使用默认内存存储和默认执行器。(默认选项分别是 MemoryJobStore 和 ThreadPoolExecutor，其中线程池的最大线程数为10)。配置完成后使用 start() 方法来启动。

如果想要显式设置 job store(使用mongo存储)和 executor 可以这样写：

# 在运行程序5秒后，第一次输出时间。
# 在 MongoDB 中可以看到 job 的状态

from datetime import datetime
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(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存储方式
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()

2.删除任务

job = scheduler.add_job(myfunc, ‘interval‘, minutes=2)
job.remove()
#如果有多个任务序列的话可以给每个任务设置ID号，可以根据ID号选择清除对象，且remove放到start前才有效
sched.add_job(myfunc, ‘interval‘, minutes=2, id=‘my_job_id‘)
sched.remove_job(‘my_job_id‘)

3. 暂停和恢复任务

4.获取job列表

获得调度任务的列表，可以使用get_jobs()来完成，它会返回所有的job实例。或者使用print_jobs()来输出所有格式化的任务列表。也可以利用get_job(任务ID)获取指定任务的任务列表

job = sched.add_job(my_job, ‘interval‘, seconds=2 ,id=‘123‘)
print sched.get_job(job_id=‘123‘)
print sched.get_jobs()

5.关闭调度器

默认情况下调度器会等待所有正在运行的任务完成后，关闭所有的调度器和任务存储。如果你不想等待，可以将wait选项设置为False。

sched.shutdown()
sched.shutdown(wait=False)

6.scheduler 事件

scheduler 可以添加事件监听器，并在特殊的时间触发。

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)

六、任务运行控制

add_job的第二个参数是trigger，它管理着任务的调度方式。它可以为date, interval或者cron。对于不同的trigger，对应的参数也相同。

1. cron定时调度（某一定时时刻执行）

# 说明

(int|str) 表示参数既可以是int类型，也可以是str类型
(datetime | str) 表示参数既可以是datetime类型，也可以是str类型

2.interval 间隔调度（每隔多久执行）

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

3.date 定时调度（作业只会执行一次）

最基本的一种调度，作业只会执行一次。它的参数如下：

原文：https://www.cnblogs.com/yblackd/p/13359769.html

评论（0）
© 2014 bubuko.com 版权所有 - 联系我们:wmxa8@hotmail.com
打开技术之扣，分享程序人生！