荐 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

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

APScheduler四个基本对象：

触发器

date触发器

interval触发器

crontab触发器

调度器

存储器

执行器

定时任务调度配置

任务操作

异常监听

相关推荐

python中bool的应用场景有哪些

怎么使用python编写简单鸡兔同笼程序

python任意进制转换的方法是什么

python怎么去掉重复数据

python列表重复元素怎么删除

python中怎么去掉重复项

python中len函数的使用方法是什么

python如何把字符串拆开