apscheduler如何在Python中实现定时？

lsekfe

前言
　　之前有介绍了用Linux crontab的方式来实现定时任务，这是使用Linux内置模块来实现的。而在Python中，还可以用第三方包来管理定时任务，比如celery、apscheduler。相对来说apscheduler使用起来更简单一些，这里来介绍一下apscheduler的使用方法。
　　首先安装起来很简单，运行pip install apscheduler即可。
　　初识apscheduler
　　来个简单的例子看看apscheduler是如何使用的。

1. <font size="3">　#encoding:utf-8
   
2. 
   
3. 　　from apscheduler.schedulers.blocking import BlockingScheduler
   
4. 
   
5. 　　import datetime
   
6. 
   
7. 　　def sch_test():
   
8. 
   
9. 　　    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
   
10. 
    
11. 　　    print('时间：{}, 测试apscheduler'.format(now))
    
12. 
    
13. 　　task = BlockingScheduler()
    
14. 
    
15. 　　task.add_job(func=sch_test, trigger='cron', second='*/10')
    
16. 
    
17. 　　task.start()</font>

复制代码

上述例子很简单，我们首先要定义一个apscheduler的对象，然后add_job添加任务，最后start开启任务就行了。
　　例子是每隔10秒运行一次sch_test任务，运行结果如下：

1. <font size="3">时间：2022-10-08 15:16:30, 测试apscheduler
   
2. 
   
3. 　　时间：2022-10-08 15:16:40, 测试apscheduler
   
4. 
   
5. 　　时间：2022-10-08 15:16:50, 测试apscheduler
   
6. 
   
7. 　　时间：2022-10-08 15:17:00, 测试apscheduler</font>

复制代码

如果我们要在执行任务函数时携带参数，只要在add_job函数中添加args就行，比如task.add_job(func=sch_test, args=('a'), trigger='cron', second='*/10')。
　　apscheduler有哪些模块
　　上面例子中我们初步了解到如何使用apschedulerl了，接下来需要知道apscheduler的设计框架。apscheduler有四个主要模块，分别是：触发器triggers、任务存储器job_stores、执行器executors、调度器schedulers。
　　1. 触发器triggers:
　　触发器指的是任务指定的触发方式，例子中我们用的是“cron”方式。我们可以选择cron、date、interval中的一个。
　　·cron表示的是定时任务，类似linux crontab，在指定的时间触发。
　　使用方法示例，在每天7点20分执行一次：
　　task.add_job(func=sch_test, args=('定时任务',), trigger='cron',
　　hour='7', minute='20')
　　·date表示具体到某个时间的一次性任务；

1. <font size="3">使用方法示例：
   
2. 
   
3. 　　# 使用run_date指定运行时间
   
4. 
   
5. 　　task.add_job(func='sch_test', trigger='date', run_date=datetime.datetime(2022 ,10 , 8, 16, 1, 30))
   
6. 
   
7. 　　# 或者用next_run_time
   
8. 
   
9. 　　task.add_job(func=sch_test,trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3))</font>

复制代码

　·interval表示的是循环任务，指定一个间隔时间，每过间隔时间执行一次。
　　使用方法示例，每隔3秒执行一次sch_test任务：
　　task.add_job(func=sch_test, args=('循环任务',), trigger='interval', seconds=3)。
　　来个例子把3种触发器都使用一遍：

1. <font size="3"># encoding:utf-8
   
2. 
   
3. 　　from apscheduler.schedulers.blocking import BlockingScheduler
   
4. 
   
5. 　　import datetime
   
6. 
   
7. 　　def sch_test(job_type):
   
8. 
   
9. 　　    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
   
10. 
    
11. 　　    print('时间：{}, {}测试apscheduler'.format(now, job_type))
    
12. 
    
13. 　　task = BlockingScheduler()
    
14. 
    
15. 　　task.add_job(func=sch_test, args=('一次性任务',),trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3))
    
16. 
    
17. 　　task.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5')
    
18. 
    
19. 　　task.add_job(func=sch_test, args=('循环任务',), trigger='interval', seconds=3)
    
20. 
    
21. 　　task.start()</font>

复制代码

打印部分结果：

1. <font size="3">时间：2022-10-08 15:45:49, 一次性任务测试apscheduler
   
2. 
   
3. 　　时间：2022-10-08 15:45:49, 循环任务测试apscheduler
   
4. 
   
5. 　　时间：2022-10-08 15:45:50, 定时任务测试apscheduler
   
6. 
   
7. 　　时间：2022-10-08 15:45:52, 循环任务测试apscheduler
   
8. 
   
9. 　　时间：2022-10-08 15:45:55, 定时任务测试apscheduler
   
10. 
    
11. 　　时间：2022-10-08 15:45:55, 循环任务测试apscheduler
    
12. 
    
13. 　　时间：2022-10-08 15:45:58, 循环任务测试apscheduler</font>

复制代码

通过代码示例和结果展示，我们可清晰的知道不同触发器的使用区别。
　　2. 任务存储器job_stores
　　顾名思义，任务存储器是存储任务的地方，默认都是存储在内存中。我们也可自定义存储方式，比如将任务存到mysql中。这里有以下几种选择：


通常默认存储在内存即可，但若程序故障重启的话，会重新拉取任务运行了，如果你对任务的执行要求高，那么可以选择其他的存储器。
　　使用SQLAlchemyJobStore存储器示例：

1. <font size="3">　from apscheduler.schedulers.blocking import BlockingScheduler
   
2. 
   
3. 　　def sch_test(job_type):
   
4. 
   
5. 　　    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
   
6. 
   
7. 　　    print('时间：{}, {}测试apscheduler'.format(now, job_type))
   
8. 
   
9. 　　sched = BlockingScheduler()
   
10. 
    
11. 　　# 使用mysql存储任务
    
12. 
    
13. 　　sql_url = 'mysql+pymysql://root:root@localhost:3306/db_name?charset=utf8'
    
14. 
    
15. 　　sched.add_jobstore('sqlalchemy',url=sql_url)
    
16. 
    
17. 　　# 添加任务
    
18. 
    
19. 　　sched.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5')
    
20. 
    
21. 　　sched.start()</font>

复制代码

3. 执行器executors
　　执行器的功能就是将任务放到线程池或进程池中运行。有以下几种选择：


默认是ThreadPoolExecutor， 常用的也就是第线程和进程池执行器。如果应用是CPU密集型操作，可用ProcessPoolExecutor来执行。
　　4. 调度器schedulers
　　调度器属于apscheduler的核心，它扮演着统筹整个apscheduler系统的角色，存储器、执行器、触发器在它的调度下正常运行。调度器有以下几个：


不是特定场景下，我们最常用的是BlockingScheduler调度器。
　　异常监听
　　定时任务在运行时，若出现错误，需要设置监听机制，我们通常结合logging模块记录错误信息。
　　使用示例：

1. <font size="3">from apscheduler.schedulers.blocking import BlockingScheduler
   
2. 
   
3. 　　import datetime
   
4. 
   
5. 　　from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR
   
6. 
   
7. 　　import logging
   
8. 
   
9. 　　# logging日志配置打印格式及保存位置
   
10. 
    
11. 　　logging.basicConfig(level=logging.INFO,
    
12. 
    
13. 　　                    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
    
14. 
    
15. 　　                    datefmt='%Y-%m-%d %H:%M:%S',
    
16. 
    
17. 　　                    filename='sche.log',
    
18. 
    
19. 　　                    filemode='a')
    
20. 
    
21. 　　def log_listen(event):
    
22. 
    
23. 　　if event.exception :
    
24. 
    
25. 　　print ( '任务出错，报错信息：{}'.format(event.exception))
    
26. 
    
27. 　　else:
    
28. 
    
29. 　　print ( '任务正常运行...' )
    
30. 
    
31. 　　def sch_test(job_type):
    
32. 
    
33. 　　    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
34. 
    
35. 　　    print('时间：{}, {}测试apscheduler'.format(now, job_type))
    
36. 
    
37. 　　    print(1/0)
    
38. 
    
39. 　　sched = BlockingScheduler()
    
40. 
    
41. 　　# 使用mysql存储任务
    
42. 
    
43. 　　sql_url = 'mysql+pymysql://root:root@localhost:3306/db?charset=utf8'
    
44. 
    
45. 　　sched.add_jobstore('sqlalchemy',url=sql_url)
    
46. 
    
47. 　　# 添加任务
    
48. 
    
49. 　　sched.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5')
    
50. 
    
51. 　　# 配置任务执行完成及错误时的监听
    
52. 
    
53. 　　sched.add_listener(log_listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
    
54. 
    
55. 　　# 配置日志监听
    
56. 
    
57. 　　sched._logger = logging
    
58. 
    
59. 　　sched.start()</font>

复制代码

apscheduler的封装使用
　　上面介绍了apscheduler框架的主要模块，我们基本能掌握怎样使用apscheduler了。下面就来封装一下apscheduler吧，以后要用直接在这份代码上修改就行了。

1. <font size="3">from apscheduler.schedulers.blocking import BlockingScheduler
   
2. 
   
3. 　　from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
   
4. 
   
5. 　　from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR
   
6. 
   
7. 　　import logging
   
8. 
   
9. 　　import logging.handlers
   
10. 
    
11. 　　import os
    
12. 
    
13. 　　import datetime
    
14. 
    
15. 　　class LoggerUtils():
    
16. 
    
17. 　　    def init_logger(self, logger_name):
    
18. 
    
19. 　　        # 日志格式
    
20. 
    
21. 　　        formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
    
22. 
    
23. 　　        log_obj = logging.getLogger(logger_name)
    
24. 
    
25. 　　        log_obj.setLevel(logging.INFO)
    
26. 
    
27. 　　        # 设置log存储位置
    
28. 
    
29. 　　        path = '/data/logs/'
    
30. 
    
31. 　　        filename = '{}{}.log'.format(path, logger_name)
    
32. 
    
33. 　　        if not os.path.exists(path):
    
34. 
    
35. 　　            os.makedirs(path)
    
36. 
    
37. 　　        # 设置日志按照时间分割
    
38. 
    
39. 　　        timeHandler = logging.handlers.TimedRotatingFileHandler(
    
40. 
    
41. 　　           filename,
    
42. 
    
43. 　　           when='D',  # 按照什么维度切割， S:秒，M：分，H:小时，D:天，W:周
    
44. 
    
45. 　　           interval=1, # 多少天切割一次
    
46. 
    
47. 　　           backupCount=10  # 保留几天
    
48. 
    
49. 　　        )
    
50. 
    
51. 　　        timeHandler.setLevel(logging.INFO)
    
52. 
    
53. 　　        timeHandler.setFormatter(formatter)
    
54. 
    
55. 　　        log_obj.addHandler(timeHandler)
    
56. 
    
57. 　　        return log_obj
    
58. 
    
59. 　　class Scheduler(LoggerUtils):
    
60. 
    
61. 　　    def __init__(self):
    
62. 
    
63. 　　        # 执行器设置
    
64. 
    
65. 　　        executors = {
    
66. 
    
67. 　　            'default': ThreadPoolExecutor(10),  # 设置一个名为“default”的ThreadPoolExecutor，其worker值为10
    
68. 
    
69. 　　            'processpool': ProcessPoolExecutor(5)  # 设置一个名为“processpool”的ProcessPoolExecutor，其worker值为5
    
70. 
    
71. 　　        }
    
72. 
    
73. 　　        self.scheduler = BlockingScheduler(timezone="Asia/Shanghai", executors=executors)
    
74. 
    
75. 　　        # 存储器设置
    
76. 
    
77. 　　        # 这里使用sqlalchemy存储器,将任务存储在mysql
    
78. 
    
79. 　　        sql_url = 'mysql+pymysql://root:root@localhost:3306/db?charset=utf8'
    
80. 
    
81. 　　        self.scheduler.add_jobstore('sqlalchemy',url=sql_url)
    
82. 
    
83. 　　        def log_listen(event):
    
84. 
    
85. 　　            if event.exception:
    
86. 
    
87. 　　                # 日志记录
    
88. 
    
89. 　　                self.scheduler._logger.error(event.traceback)
    
90. 
    
91. 　　   
    
92. 
    
93. 　　        # 配置任务执行完成及错误时的监听
    
94. 
    
95. 　　        self.scheduler.add_listener(log_listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
    
96. 
    
97. 　　        # 配置日志监听
    
98. 
    
99. 　　        self.scheduler._logger = self.init_logger('sche_test')
    
100. 
     
101. 　　    def add_job(self, *args, **kwargs):
     
102. 
     
103. 　　        """添加任务"""
     
104. 
     
105. 　　        self.scheduler.add_job(*args, **kwargs)
     
106. 
     
107. 　　    def start(self):
     
108. 
     
109. 　　        """开启任务"""
     
110. 
     
111. 　　        self.scheduler.start()
     
112. 
     
113. 　　# 测试任务
     
114. 
     
115. 　　def sch_test(job_type):
     
116. 
     
117. 　　    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
     
118. 
     
119. 　　    print('时间：{}, {}测试apscheduler'.format(now, job_type))
     
120. 
     
121. 　　    print(1/0)
     
122. 
     
123. 　　# 添加任务,开启任务
     
124. 
     
125. 　　sched = Scheduler()
     
126. 
     
127. 　　# 添加任务
     
128. 
     
129. 　　sched.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5')
     
130. 
     
131. 　　# 开启任务
     
132. 
     
133. 　　sched.start()</font>

复制代码

小结
　　这篇文章介绍了Python实现定时任务的又一利器apscheduler，通过简单例子及apscheduler框架的主要模块分解，我们可以根据实际需求配置好模块信息，再结合logging模块，我们可以实时监控到定时任务的运行情况。话说大家是如何来实现定时任务的，欢迎评论区讨论。