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

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("https://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("https://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("https://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：任务触发的误差时间。

也可以用表达式类型，可以用以下方式：

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("https://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("https://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()