北京地区做网站推广用哪家的好自动点击器
前言：本博客仅作记录学习使用，部分图片出自网络，如有侵犯您的权益，请联系删除
目录
定时任务模块APScheduler
一、安装及基本概念
1.1、APScheduler的安装
1.2、涉及概念
1.3、APScheduler的工作流程编辑
二、配置调度器
三、启动调度器
四、调度事件监听
定时任务模块APScheduler
APScheduler提供了基于日期、固定时间间隔以及crontab类型的任务，我们可以在主程序的运行过程中快速增加新作业或删除旧作业。如果把作业存储在数据库中，那么作业的状态会被保存，当调度器重启时，不必重新添加作业，作业会恢复原状态继续执行。
一、安装及基本概念
1.1、APScheduler的安装
pip install apscheduler
1.2、涉及概念
- 触发器（triggers）：触发器包含调度逻辑，描述一个任务何时被触发，有按日期、按时间间隔、按cronjob描述式三种触发方式。每个作业都有自己的触发器，除了初始配置之外，触发器是完全无状态的。
- 作业存储器（job stores）：指定了作业被存放的位置，默认的作业存储器是内存，也可以将作业保存在各种数据库中。当作业被存放在数据库中时，它会被序列化；当重新被加载时，会反序列化。作业存储器充当保存、加载、更新和查找作业的中间商。在调度器之间不能共享作业存储
- 执行器（executors）：执行器是将指定的作业（调用函数）提交到线程池或进程池中运行，当任务完成时，执行器通知调度器触发相应的事件。
- 调度器（schedulers）：任务调度器，控制器角色，通过它配置作业存储器、执行器和触发器、添加、修改和删除任务。调度器协调触发器、作业存储器、执行器的运行，通常只有一个调度程序运行在应用程序中，开发人员不需要直接处理作业存储器、执行器或触发器。配置作业存储器和执行器是通过调度器来完成的
1.3、APScheduler的工作流程
一个简单的间隔任务实例：
import osfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingScheduler# 打印当前的时间def tick():print('Tick! The time is: %s' % datetime.now())if __name__ == '__main__':scheduler = BlockingScheduler()# 添加一个作业rick，触发器为interval，每隔3秒执行一次scheduler.add_job(tick, 'interval', seconds=3)print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass另外的触发器为date，cron。date按特定时间点触发，cron则按固定的时间间隔触发。
上述代码稍作修改可变为cron类的定时任务：
import osfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingSchedulerdef tick():print('Tick! The time is: %s' % datetime.now())if __name__ == '__main__':scheduler = BlockingScheduler()scheduler.add_job(tick, 'cron', hour=19,minute=23)print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass定时cron任务也非常简单，直接给触发器trigger传入'cron'即可。hour=19，minute23，表示每天的19时23分执行任务
hour=19,minute=23hour='19',minute='23'minute='*/3' # 表示每3分钟执行一次hour='19-21',minute='23' # 表示19:23、20:23、21:23各执行一次任务二、配置调度器
调度器的主循环其实就是反复检查是否有到期需要执行的任务，具体分两步进行
- 询问自己的每一个作业存储器，有没有到期需要执行的任务。如果有则计算这些作业中每个作业需要 运行的时间点；如果时间点有多个，就做coalesce检查。
-
提交给执行器按时间点运行
各调度器的适用场景：
- BlockingSchduler：适用于调度程序，是进程中唯一运行的进程，调用start函数会阻塞当前线程，不能立即返回
- BackgroundScheduler：适用于调度程序，在应用程序的后台运行，调用start后主线程不会阻塞。
- AsyncIOScheduler：适用于使用了asyncio模块的应用程序
- GeventScheduler：适用于使用了gevent模块的应用程序
- TwistedScheduler：适用于构建Twisted的应用程序
- QtSchuduler：适用于构建Qt的应用程序。
作业存储器的选择：一是内存（ 默认），而是数据库。
执行器的选择：默认的ThreadPoolExecutor足够OK，如果作业负载涉及CPU密集型操作，那么考虑使用ProcessPoolExecutor，甚至同时使用，将其作为二级执行器。
APScheduler可以使用字典，关键字参数传递配置调度器。首先实例化调度程序添加作业，然后配置调度器，获得最大的灵活性。
如果调度程序在应用程序的后台运行，则选择BackgroundScheduler，并使用默认的jobstore和executor
from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = BlockingScheduler()
如果想配置更多的信息，就可设置两个执行器、两个作业存储器、调整新作业的默认值，并设置不同的时区。配置详情：
- 配置名为mongo的MongoDBjobStore作业存储器
- 配置名为default的SQLAlchemyJobStore（使用SQLite）
- 配置名为default的ThreadPoolExecutor，最大进程数为5
- UTC作为调度器的时区
- coalesce默认情况下关闭
- 作业的默认最大运行实例限制为3
方法一：
from pytz import utcfrom apscheduler.schedulers.background import BlockingSchedulerfrom apscheduler.jobstores.mongodb import MongoDBJobStorefrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorjobstores = {'mongo':MongoDBJobStore(),'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(5)}job_defaults = {'coalesce':False,'max_instances':3}scheduler = BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)方法二：
from apscheduler.schedulers.background import BlockingSchedulerscheduler = BlockingScheduler({'apscheduler.jobstores.mongo':{'type':'mongodb'},'apscheduler.jobstores.default':{'type':'sqlalchemy','url':'sqlite:///jobs.sqlite'},'apscheduler.executors.default':{'class':'apscheduler.executors.pool:ThreadPoolExecutor','max_workers':'5'},'apscheduler.job_defaults.coalesce':'fasle','apscheduler.job_defaults.max_instances':'3','apscheduler.timezone':'UTC',})方法三：
from pytz import utcfrom apscheduler.schedulers.background import BlockingSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorjobstores = {'mongo':{'type':'mongodb'},'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':{'type':'threadpool','max_workers':20},'processpool':ProcessPoolExecutor(max_workers=5)}job_defaults = {'coalesce':False,'max_instances':3}scheduler = BlockingScheduler()scheduler.configure(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)三、启动调度器
启动调度器前需要先添加作业，有两种方法可以向调度器添加作业：一是通过接口add_job()；二是通过使用函数装饰器，其中add_job()返回一个apscheduler.job.Job类的实例，用于后续修改或删除作业。
可以随时在调度器上调度作业。如果在添加作业时，调度器还没有启动，那么任务不会运行，并且它的第一次运行时间在调度器启动时计算。
调用调度器的start()方法启动调度器，下面用不同的作业存储器来举例：
from apscheduler.schedulers.blocking import BlockingSchedulerimport datetimefrom apscheduler.jobstores.memory import MemoryJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorfdef my_job(id='my_job'):print(id,'-->',datetime.datetime.now())jobstores = {'default':MemoryJobStore()}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(10)}job_defaults = {'coalesce':False,'max_instance':3}scheduler =BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults)scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',end_date='2024-06-06')scheduler.add_job(my_job,args=['job_once_now',],id='job_once_now')scheduler.add_job(my_job,args=['job_date_once',],id='job_date_once',trigger='date',run_date='2024-01-01 00:00:00')try:scheduler.start()except SystemExit:print('exit')exit()方法二：使用数据库作为作业存储器（修改第5行和11行）
from apscheduler.schedulers.blocking import BlockingSchedulerimport datetimefrom apscheduler.jobstores.memory import MemoryJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStoredef my_job(id='my_job'):print(id,'-->',datetime.datetime.now())jobstores = {'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(10)}job_defaults = {'coalesce':False,'max_instance':3}scheduler =BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults)scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',end_date='2024-06-06')scheduler.add_job(my_job,args=['job_once_now',],id='job_once_now')scheduler.add_job(my_job,args=['job_date_once',],id='job_date_once',trigger='date',run_date='2024-01-01 00:00:00')try:scheduler.start()except SystemExit:print('exit')exit()运行过之后，如果不注释添加作业的代码，则作业会重新添加到数据库中，这样就有了两个作业，为了避免这样的情况：设置（replace_existing=True）
scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)如果想运行错过运行的作业，则使用misfire_grace_time：
scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',coalesce=True,misfire_grace_time=30,replace_existing=True,end_date='2024-06-06')其他操作如下：
scheduler.remove_job(job_id,jobstore=None) # 删除作业scheduler.remove_all_jobs(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 BlockingSchedulerfrom apscheduler.events import EVENT_JOB_EXECUTED,EVENT_JOB_ERRORimport datetimeimport 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 aps_test(x):print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),x)def date_test(x):print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),x)print(1/0)def my_listener(event):if event.exception:print('任务出错了!!!!')else:print('任务照常运行...')scheduler = BlockingScheduler()scheduler.add_job(func=date_test,args=('一次性任务，会出错',),next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=15),id='date_task')scheduler.add_job(func=aps_test,args=('循环任务',),trigger='interval',seconds=3,id='interval_task')scheduler.add_listener(my_listener,EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)scheduler._logger = loggingscheduler.start()致谢
在此，我要对所有为知识共享做出贡献的个人和机构表示最深切的感谢。同时也感谢每一位花时间阅读这篇文章的读者，如果文章中有任何错误，欢迎留言批评指正。
学习永无止境，让我们共同进步