上一篇日志我们学习了APSchedule定时任务框架的大致用法，现在再来实战练习下。

实例1：最简单的例子

每个三秒输出次tick

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import os
from apscheduler.schedulers.blocking import BlockingScheduler
if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_job('sys:stdout.write', 'interval', seconds=3, args=['tick\n'])
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

实例2：使用ProcessPoolExecutor

这个例子中我们将任务放在进程池中，这个任务每隔三秒输出一次当前时间：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
"""
Demonstrates how to schedule a job to be run in a process pool on 3 second intervals.
"""
from datetime import datetime
import os
from apscheduler.schedulers.blocking import BlockingScheduler
def tick():
    print('Tick! The time is: %s' % datetime.now())
if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_executor('processpool')
    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

实例3：使用任务仓库

下面的例子使用SQLAlchemy任务仓库，每次执行执行任务时，都会添加一个新的alarm任务，你可以退出程序，然后重启，查看之前没有执行的alarm，也可以使用数据库的url作为参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
"""
This example demonstrates the use of the SQLAlchemy job store.
On each run, it adds a new alarm that fires after ten seconds.
You can exit the program, restart it and observe that any previous alarms that have not fired yet
are still active. You can also give it the database URL as an argument.
See the SQLAlchemy documentation on how to construct those.
"""
from datetime import datetime, timedelta
import sys
import os
from apscheduler.schedulers.blocking import BlockingScheduler
def alarm(time):
    print('Alarm! This alarm was scheduled at %s.' % time)
if __name__ == '__main__':
    scheduler = BlockingScheduler()
    url = sys.argv[1] if len(sys.argv) > 1 else 'sqlite:///example.sqlite'
    scheduler.add_jobstore('sqlalchemy', url=url)
    alarm_time = datetime.now() + timedelta(seconds=10)
    scheduler.add_job(alarm, 'date', run_date=alarm_time, args=[datetime.now()])
    print('To clear the alarms, delete the example.sqlite file.')
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass