python apscheduler的使用

本文主要是介绍python apscheduler的使用，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

# !/usr/bin/env python3
# -*- coding: utf-8 -*-

import pytz
from util import log, cfg
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.schedulers.blocking import BlockingScheduler
logger = log.get_logger('timed_task', 'INFO')


def init():
    logger.info("scheduler_handler main start run")


def test():
    print('测试成功')


def main():
    # 执行需要初始化数据的处理
    init()

    # 时区设定
    timezone = pytz.timezone('Asia/Shanghai')

    # 执行器
    executors = {'default': ThreadPoolExecutor(20), 'processpool': ProcessPoolExecutor(3)}
    job_defaults = {'coalesce': True, 'max_instances': 3}
    # 阻塞主线程的运行
    scheduler = BlockingScheduler(timezone=timezone, executors=executors, job_defaults=job_defaults)

    ####################################################################################################################

    if cfg.sys == 'Linux':
        scheduler.add_job(func=test, trigger='interval', seconds=10)
        scheduler.add_job(func=test, trigger="cron", day_of_week='*', hour='23', minute='50')

    else:
        scheduler.add_job(func=test, trigger='interval', seconds=10)

    ####################################################################################################################
    def event_listener(event):

        if event.exception:
            exception = str(event.exception)
            scheduled_run_time = str(event.scheduled_run_time)
            traceback = str(event.traceback)

            # 报错信息打印出来
            print("\n\n", "#" * 50, "\n\n")
            logger.info(
                f"event scheduled_run_time:{scheduled_run_time} \n exception: {exception} \n traceback: {traceback}")
            print("\n\n", "#" * 50, "\n\n")
        else:
            # 正常执行的处理
            pass
    #
    # # 监听正常运行和运行异常
    scheduler.add_listener(event_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

    try:
        # 启动定时器
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        logger.debug("scheduler start-up fail")


if __name__ == '__main__':
    main()

这篇关于python apscheduler的使用的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！