Python快速而美丽[v1.0.0][定时任务]

2020/7/14 15:29:20 人评论 次浏览 分类:学习教程

Python中定时任务的解决方案,总体来说就四种,分别是:crontab、 scheduler、 Celery、 APScheduler,其中 crontab不适合多台服务器的配置、scheduler太过于简单、 Celery依赖的软件比较多,比较耗资源。最好的解决方案就是 APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
from selenium import webdriver
from time import sleep


def schedule_task():
    """
    create one schedule visitor
    :return:
    """
    chrome_driver = webdriver.Chrome()
    chrome_driver.get("http://online10110.leadscloud.com")
    chrome_driver.maximize_window()
    sleep(3)
    chrome_driver.find_element_by_id("xhl-launch-button-div").click()
    sleep(3)
    chrome_driver.find_element_by_id("messageText").send_keys("robot visitor")
    sleep(3)
    chrome_driver.find_element_by_id("sendBtn").click()
    print("current time is: %s" % datetime.now())


if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_job(schedule_task, "interval", seconds=3)
    scheduler.start()

APScheduler四个基本对象:

  • 触发器(triggers): 触发器就是根据你指定的触发方式,比如是按照时间间隔,还是按照crontab触发,触发条件是什么等。每个任务都有自己的触发器
  • 任务存储器(job stores):任务存储器是可以存储任务的地方,默认情况下任务保存在内存,也可将任务保存在各种数据库中。任务存储进去后,会进行序列化,然后也可以反序列化提取出来,继续执行
  • 执行器(executors): 执行器的目的是安排任务到线程池或者进程池中运行的
  • 调度器(schedulers):任务调度器是属于整个调度的总指挥官。他会合理安排作业存储器、执行器、触发器进行工作,并进行添加和删除任务等。调度器通常是只有一个的。开发人员很少直接操作触发器、存储器、执行器等。因为这些都由调度器自动来实现了

触发器

date触发器

在某个日期时间只触发一次事件


from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
from selenium import webdriver
from time import sleep
from datetime import datetime


def schedule_task():
    """
    create one schedule visitor
    :return:
    """
    chrome_driver = webdriver.Chrome()
    chrome_driver.get("http://online10110.leadscloud.com")
    chrome_driver.maximize_window()
    sleep(3)
    chrome_driver.find_element_by_id("xhl-launch-button-div").click()
    sleep(3)
    chrome_driver.find_element_by_id("messageText").send_keys("robot visitor")
    sleep(3)
    chrome_driver.find_element_by_id("sendBtn").click()
    print("current time is: %s" % datetime.now())


if __name__ == '__main__':
    scheduled_job = BlockingScheduler()
    scheduled_job.add_job(schedule_task, 'date', run_date=date(2020, 5, 22))
    scheduled_job.start()

interval触发器

想要在固定的时间间隔触发事件。

interval的触发器可以设置以下的触发参数:

  • weeks:周。整形
  • days:一个月中的第几天,整型
  • hours:小时,整形
  • minutes:分钟,整形
  • seconds:秒,整形
  • start_date:间隔触发的起始时间
  • end_date:间隔触发的结束时间
  • jitter:触发的时间误差
from apscheduler.schedulers.blocking import BlockingScheduler
from selenium import webdriver
from time import sleep
from datetime import datetime


def schedule_task():
   """
   create one schedule visitor
   :return:
   """
   chrome_driver = webdriver.Chrome()
   chrome_driver.get("http://online10110.leadscloud.com")
   chrome_driver.maximize_window()
   sleep(3)
   chrome_driver.find_element_by_id("xhl-launch-button-div").click()
   sleep(3)
   chrome_driver.find_element_by_id("messageText").send_keys("robot visitor")
   sleep(3)
   chrome_driver.find_element_by_id("sendBtn").click()
   print("current time is: %s" % datetime.now())


if __name__ == '__main__':
   scheduled_job = BlockingScheduler()
   scheduled_job.add_job(schedule_task, 'interval', hours=2)
   scheduled_job.start()

还可以加上起止时间scheduled_job.add_job(schedule_task, 'interval', hours=2, start_date='2020-10-10 09:30:00', end_date='2024-06-15 11:00:00')

crontab触发器

在某个确切的时间周期性的触发事件

  • year:4位数字的年份
  • month:1-12月份
  • day:1-31日
  • week:1-53周
  • day_of_week:一个礼拜中的第几天( 0-6或者 mon、 tue、 wed、 thu、 fri、 sat、 sun)
  • hour: 0-23小时
  • minute: 0-59分钟
  • second: 0-59秒
  • start_date: datetime类型或者字符串类型,起始时间
  • end_date: datetime类型或者字符串类型,结束时间
  • timezone:时区
  • jitter:任务触发的误差时间。

也可以用表达式类型,可以用以下方式:

表达式 字段 描述
* 任何 在每个值都触发
*/a 任何 每隔 a触发一次
a-b 任何 在 a-b区间内任何一个时间触发( a必须小于 b)
a-b/c 任何 在 a-b区间内每隔 c触发一次
xth y day 第 x个星期 y触发
lastx day 最后一个星期 x触发
last day 一个月中的最后一天触发
x,y,z 任何 可以把上面的表达式进行组合

month和day_of_week两个字段接受英文的月和星期的缩写,例如jan – dec and mon – sun

from apscheduler.schedulers.blocking import BlockingScheduler
from selenium import webdriver
from time import sleep
from datetime import datetime


def schedule_task():
    """
    create one schedule visitor
    :return:
    """
    chrome_driver = webdriver.Chrome()
    chrome_driver.get("http://online10110.leadscloud.com")
    chrome_driver.maximize_window()
    sleep(3)
    chrome_driver.find_element_by_id("xhl-launch-button-div").click()
    sleep(3)
    chrome_driver.find_element_by_id("messageText").send_keys("robot visitor")
    sleep(3)
    chrome_driver.find_element_by_id("sendBtn").click()
    print("current time is: %s" % datetime.now())


if __name__ == '__main__':
    schedule_job = BlockingScheduler()
    schedule_job.add_job(schedule_task, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
    schedule_job.start()

调度器

  • BlockingScheduler:适用于调度程序是进程中唯一运行的进程,调用 start函数会阻塞当前线程,不能立即返回
  • BackgroundScheduler:适用于调度程序在应用程序的后台运行,调用 start后主线程不会阻塞
  • AsyncIOScheduler:适用于使用了 asyncio模块的应用程序
  • GeventScheduler:适用于使用 gevent模块的应用程序
  • TwistedScheduler:适用于构建 Twisted的应用程序
  • QtScheduler:适用于构建 Qt的应用程序

存储器

任务存储器的选择有两种。一是内存,也是默认的配置。二是数据库。使用内存的方式是简单高效,但是不好的是,一旦程序出现问题,重新运行的话,会把之前已经执行了的任务重新执行一遍。数据库则可以在程序崩溃后,重新运行可以从之前中断的地方恢复正常运行

  • MemoryJobStore:没有序列化,任务存储在内存中,增删改查都是在内存中完成
  • SQLAlchemyJobStore:使用 SQLAlchemy这个 ORM框架作为存储方式
  • MongoDBJobStore:使用 mongodb作为存储器。
  • RedisJobStore:使用 redis作为存储器。

执行器

执行器的选择取决于应用场景。通常默认的 ThreadPoolExecutor已经在大部分情况下是可以满足我们需求的。如果我们的任务涉及到一些 CPU密集计算的操作。那么应该考虑 ProcessPoolExecutor。然后针对每种程序, apscheduler也设置了不同的 executor

  • ThreadPoolExecutor:线程池执行器
  • ProcessPoolExecutor:进程池执行器
  • GeventExecutor:Gevent程序执行器
  • TornadoExecutor:Tornado程序执行器
  • TwistedExecutor:Twisted程序执行器
  • AsyncIOExecutor: asyncio程序执行器

定时任务调度配置

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from selenium import webdriver
from time import sleep


def schedule_task():
    """
    create one schedule visitor
    :return:
    """
    chrome_driver = webdriver.Chrome()
    chrome_driver.get("http://online10110.leadscloud.com")
    chrome_driver.maximize_window()
    sleep(3)
    chrome_driver.find_element_by_id("xhl-launch-button-div").click()
    sleep(3)
    chrome_driver.find_element_by_id("messageText").send_keys("robot visitor")
    sleep(3)
    chrome_driver.find_element_by_id("sendBtn").click()
    print("current time is: %s" % datetime.now())


def interval_task():
    # 配置default的任务存储器为SQLAlchemyJobStore(使用SQLite)
    jobstores={'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}
    # 配置 default执行器为 ThreadPoolExecutor,并且设置最多的线程数是20个
    executors={'default':ThreadPoolExecutor(20)}
    '''
    设置 coalesce为 False:设置这个目的是,比如由于某个原因导致某个任务积攒了很多次没有执行(比如有一个任务是1分钟跑一次,但是系统原因断了5分钟)
    如果 coalesce=True,那么下次恢复运行的时候,会只执行一次,而如果设置 coalesce=False,那么就不会合并,会5次全部执行。
    max_instances=5:同一个任务同一时间最多只能有5个实例在运行。比如一个耗时10分钟的job,被指定每分钟运行1次,如果我 max_instance值5,
    那么在第6~10分钟上,新的运行实例不会被执行,因为已经有5个实例在跑了
    '''
    job_defaults={'coalesce':False,'max_instances':3}
    scheduler=BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
    scheduler.add_job(schedule_task, "interval", minutes=1)
    scheduler.start()


if __name__ == '__main__':
    interval_task()

任务操作

  • 添加任务:使用 scheduler.add_job(job_obj,args,id,trigger,**trigger_kwargs)
  • 删除任务:使用 scheduler.remove_job(job_id,jobstore=None)
  • 暂停任务:使用 scheduler.pause_job(job_id,jobstore=None)
  • 恢复任务:使用 scheduler.resume_job(job_id,jobstore=None)
  • 修改某个任务属性信息:使用 scheduler.modify_job(job_id,jobstore=None,**changes)
  • 修改单个作业的触发器并更新下次运行时间:使用 scheduler.reschedule_job(job_id,jobstore=None,trigger=None,**trigger_args)
  • 输出作业信息:使用 scheduler.print_jobs(jobstore=None,out=sys.stdout)

异常监听

当我们的任务抛出异常后,我们可以监听到,然后把错误信息进行记录。示例代码如下:

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 one_time_task(say_something):
    """
    一次性任务
    :param say_something: 
    :return: 
    """
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), say_something)


def loop_task(say_something):
    """
    循环任务
    :param say_something: 
    :return: 
    """
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), say_something)
    # 故意抛出异常
    print(1 / 0)


def schedule_job_listener(event):
    """
    定时任务监听器
    :param event: 
    :return: 
    """
    if event.exception:
        print('任务出错了!!!!!!')
    else:
        print('任务照常运行......')


scheduler = BlockingScheduler()
scheduler.add_job(func=one_time_task, args=('一次性任务,会出错',),
                  next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=15), id='date_task')
scheduler.add_job(func=loop_task, args=('循环任务',), trigger='interval', seconds=3, id='interval_task')

# 配置任务执行完成和执行错误的监听
scheduler.add_listener(schedule_job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

# 设置日志
scheduler._logger = logging
scheduler.start()

相关资讯

    暂无相关的资讯...

共有访客发表了评论 网友评论

验证码: 看不清楚?
    -->