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

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

本文地址：https://blog.csdn.net/dawei_yang000000/article/details/107334753