python定时任务模块APScheduler

一、简单任务

定义一个函数,然后定义一个scheduler类型,添加一个job,然后执行,就可以了

5秒整倍数,就执行这个函数

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def aps_test():
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), ‘你好‘)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, trigger=‘cron‘, second=‘*/5‘)
scheduler.start()

带参数的

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def aps_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=(‘你好‘,), trigger=‘cron‘, second=‘*/5‘)
scheduler.start()
# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def aps_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=(‘定时任务‘,), trigger=‘cron‘, second=‘*/5‘)
scheduler.add_job(func=aps_test, args=(‘一次性任务‘,), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12))
scheduler.add_job(func=aps_test, args=(‘循环任务‘,), trigger=‘interval‘, seconds=3)

scheduler.start()

二、日志

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import logging

logging.basicConfig(level=logging.INFO,
                    format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘,
                    datefmt=‘%Y-%m-%d %H:%M:%S‘,
                    filename=‘log1.txt‘,
                    filemode=‘a‘)

def aps_test(x):
    print 1/0
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=(‘定时任务‘,), trigger=‘cron‘, second=‘*/5‘)
scheduler._logger = logging
scheduler.start()

三、删除任务

要求执行一定阶段任务以后,删除某一个循环任务,其他任务照常进行。有如下代码:

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import logging

logging.basicConfig(level=logging.INFO,
                    format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘,
                    datefmt=‘%Y-%m-%d %H:%M:%S‘,
                    filename=‘log1.txt‘,
                    filemode=‘a‘)

def aps_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

def aps_date(x):
    scheduler.remove_job(‘interval_task‘)
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=(‘定时任务‘,), trigger=‘cron‘, second=‘*/5‘, id=‘cron_task‘)
scheduler.add_job(func=aps_date, args=(‘一次性任务,删除循环任务‘,), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12), id=‘date_task‘)
scheduler.add_job(func=aps_test, args=(‘循环任务‘,), trigger=‘interval‘, seconds=3, id=‘interval_task‘)
scheduler._logger = logging

scheduler.start()

四、停止任务,恢复任务

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import logging

logging.basicConfig(level=logging.INFO,
                    format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘,
                    datefmt=‘%Y-%m-%d %H:%M:%S‘,
                    filename=‘log1.txt‘,
                    filemode=‘a‘)

def aps_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

def aps_pause(x):
    scheduler.pause_job(‘interval_task‘)
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

def aps_resume(x):
    scheduler.resume_job(‘interval_task‘)
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=(‘定时任务‘,), trigger=‘cron‘, second=‘*/5‘, id=‘cron_task‘)
scheduler.add_job(func=aps_pause, args=(‘一次性任务,停止循环任务‘,), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12), id=‘pause_task‘)
scheduler.add_job(func=aps_resume, args=(‘一次性任务,恢复循环任务‘,), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=24), id=‘resume_task‘)
scheduler.add_job(func=aps_test, args=(‘循环任务‘,), trigger=‘interval‘, seconds=3, id=‘interval_task‘)
scheduler._logger = logging

scheduler.start()

五、捕获错误

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
import datetime
import logging

logging.basicConfig(level=logging.INFO,
                    format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘,
                    datefmt=‘%Y-%m-%d %H:%M:%S‘,
                    filename=‘log1.txt‘,
                    filemode=‘a‘)

def aps_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)

def date_test(x):
    print (datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘), x)
    print (1/0)

def my_listener(event):
    if event.exception:
        print (‘任务出错了!!!!!!‘)
    else:
        print (‘任务照常运行...‘)

scheduler = BlockingScheduler()
scheduler.add_job(func=date_test, args=(‘一定性任务,会出错‘,), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=15), id=‘date_task‘)
scheduler.add_job(func=aps_test, args=(‘循环任务‘,), trigger=‘interval‘, seconds=3, id=‘interval_task‘)
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
scheduler._logger = logging

scheduler.start()

原文地址:https://www.cnblogs.com/angelyan/p/11297252.html

时间: 2024-10-12 02:31:12

python定时任务模块APScheduler的相关文章

[转]Python定时任务框架APScheduler

APScheduler是基于Quartz的 一个Python定时任务框架,实现了Quartz的所有功能,使用起来十分方便.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以 持久化任务.基于这些功能,我们可以很方便的实现一个python定时任务系统,写python还是要比java舒服多了. 1. 安装 安装过程很简单,可以基于easy_install和源码. easy_install apscheduler 或者下载源码,运行命令: python setup.py install

Python定时任务框架APScheduler 3.0.3 Cron示例

APScheduler是基于Quartz的一个Python定时任务框架,实现了Quartz的所有功能,使用起来十分方便.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以持久化任务.基于这些功能,我们可以很方便的实现一个python定时任务系统. 安装 安装过程很简单,可以基于pip和源码. Pip install apscheduler==3.0.3 或者下载源码,运行命令: Python setup.py install cron job例子 1: #coding=utf-8

python 定时任务 from apscheduler.schedulers.blocking import BlockingScheduler

说明:使用python内置的模块来实现,本篇博客只是以循环定时来示范,其他的可以结合crontab的风格自己设定 一.导包 from apscheduler.schedulers.blocking import BlockingScheduler二.普通函数的使用 (1)interval模式,功能比较单一 sched = BlockingScheduler()@sched.scheduled_job('interval',seconds=5)def crawl_job(): #五秒执行一次 pr

定时任务框架APScheduler学习详解

APScheduler简介 在平常的工作中几乎有一半的功能模块都需要定时任务来推动,例如项目中有一个定时统计程序,定时爬出网站的URL程序,定时检测钓鱼网站的程序等等,都涉及到了关于定时任务的问题,第一时间想到的是利用time模块的time.sleep()方法使程序休眠来达到定时任务的目的,虽然这样也可以,但是总觉得不是那么的专业,^_^所以就找到了python的定时任务模块APScheduler: APScheduler基于Quartz的一个Python定时任务框架,实现了Quartz的所有功

Python下定时任务框架APScheduler的使用

1.APScheduler简介: APScheduler是Python的一个定时任务框架,可以很方便的满足用户定时执行或者周期执行任务的需求,它提供了基于日期date.固定时间间隔interval .以及类似于Linux上的定时任务crontab类型的定时任务.并且该框架不仅可以添加.删除定时任务,还可以将任务存储到数据库中,实现任务的持久化,所以使用起来非常方便. 2.APScheduler安装: APScheduler的安装相对来说也非常简单,可以直接利用pip安装,如果没有pip可以下载源

Python任务调度模块APScheduler

目录 一.APScheduler 是什么&APScheduler四种组成部分? 1.调度器(scheduler) 2.作业存储(job store) 3.执行器(executor) 4.触发器(triggers) 4.1.date 定时调度(作业只会执行一次) 4.2.interval: 每隔一段时间执行一次 4.3.cron: 使用同linux下crontab的方式 二. How:APSched 怎么用? 安装 快速上手 任务操作 1.添加任务 2.删除任务 3.暂停&继续任务 4.修改

Python任务调度模块 – APScheduler

APScheduler是一个Python定时任务框架,使用起来十分方便.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以持久化任务.并以daemon方式运行应用.目前最新版本为3.0.x. 在APScheduler中有四个组件: 触发器(trigger)包含调度逻辑,每一个作业有它自己的触发器,用于决定接下来哪一个作业会运行.除了他们自己初始配置意外,触发器完全是无状态的. 作业存储(job store)存储被调度的作业,默认的作业存储是简单地把作业保存在内存中,其他的作业存储

Python 定时任务设计

一,当你想让你某个方法在几秒甚至更长的时间内执行后执行一次,你可以这样做: import time from threading import Timer def print_time( enter_time ): print "now is", time.time() , "enter_the_box_time is", enter_time print time.time() Timer(5, print_time, ( time.time(), )).star

Python任务调度模块 – APScheduler(摘抄)

Python任务调度模块 – APScheduler 2015年6月11日 by debugo · 32 Comments APScheduler是一个Python定时任务框架,使用起来十分方便.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以持久化任务.并以daemon方式运行应用.目前最新版本为3.0.x.在APScheduler中有四个组件:触发器(trigger)包含调度逻辑,每一个作业有它自己的触发器,用于决定接下来哪一个作业会运行.除了他们自己初始配置意外,触发器完