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()

六、定时任务的接口设计

# 定时任务功能

@admin.route('/pause', methods=['POST'])

@user_login_req

def pausetask():  # 停止

    data = request.form.get('task_id')

    job = scheduler.get_job(str(data))

    res = {}

    if job:

        if 'pause' in job.__str__():

            res.update({'status': 1001, 'msg': '已停止'})

        else:

            scheduler.pause_job(str(data))

            res.update({'status': 1000, 'msg': '停止中'})

    else:

        res.update({'status': 1001, 'msg': '未运行'})

    return jsonify(res)

@admin.route('/resume', methods=['POST'])

@user_login_req

def resumetask():  # 恢复

    data = request.form.get('task_id')

    job=scheduler.get_job(str(data))

    res={}

    if job:

        if 'run' in job.__str__():

            res.update({'status':1001,'msg':'已恢复'})

        else:

            scheduler.resume_job(str(data))

            res.update({'status': 1000, 'msg': '恢复中'})

    else:

        res.update({'status':1001,'msg':'未运行'})

    return jsonify(res)

@admin.route('/remove_task', methods=['POST'])

@user_login_req

def remove_task():  # 移除

    data = request.form['task_id']

    job = scheduler.get_job(str(data))

    res = {}

    if not job:

        res.update({'status': 1001, 'msg': '已删除'})

    else:

        scheduler.remove_job(str(data))

        res.update({'status': 1000, 'msg': '删除中'})

    return jsonify(res)

@admin.route('/addjob', methods=['POST'])

@user_login_req

def addtask():

    data = request.form.get('task_id')

    job = scheduler.get_job(str(data))

    if job:

        return jsonify({'status': 1001,'msg':'已开启'})

    if data == '':

        scheduler.add_job(func=task1, id='', trigger='cron', day_of_week='0-6', hour=18, minute=19,

                          second=10,

                          replace_existing=True)

        # trigger='cron' 表示是一个定时任务

    else:

        scheduler.add_job(func=task2, id='', trigger='interval', seconds=10,

                          replace_existing=True)

        # trigger='interval' 表示是一个循环任务，每隔多久执行一次

    return jsonify({'status': 1000,'msg':'运行中'})

def task1():

    print('mession1')

    print(datetime.datetime.now())

def task2():

    print('mession2')

    print(datetime.datetime.now())

巴特西

python定时任务模块APScheduler

一、简单任务

二、日志

三、删除任务

四、停止任务，恢复任务

五、捕获错误

六、定时任务的接口设计

最新文章

热门文章