apscheduler的使用

  最近一個程序要用到後臺定時任務,看了看python後臺任務,通常2個選擇,一個是apscheduler,一個celery。apscheduler比較直觀簡單一點,就選說說這個庫吧。網上一搜索,暈死,好多寫apscheduler的都是超級老的版本,並且博客之間相互亂抄,錯誤一大堆。仍是本身讀官方文檔,爲你們理一遍吧。python

  先安裝一下吧,最新版本的apscheduler是3.0.5版web

  1. 安裝  
pip install apscheduler

  

  安裝完畢安全

  2. 簡單任務異步

  首先,來個最簡單的例子,看看它的威力。ide

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 
 5 
 6 def aps_test():
 7     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), '你好'
 8 
 9 
10 scheduler = BlockingScheduler()
11 scheduler.add_job(func=aps_test, trigger='cron', second='*/5')
12 scheduler.start()

  看代碼,定義一個函數,而後定義一個scheduler類型,添加一個job,而後執行,就能夠了,代碼是否是超級簡單,並且很是清晰。看看結果吧。函數

5秒整倍數,就執行這個函數,是否是超級超級簡單?對了,apscheduler就是通俗易懂。this

再寫一個帶參數的。spa

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 
 5 
 6 def aps_test(x):
 7     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
 8 
 9 scheduler = BlockingScheduler()
10 scheduler.add_job(func=aps_test, args=('你好',), trigger='cron', second='*/5')
11 scheduler.start()

結果跟上面同樣的。3d

好了,上面只是給你們看的小例子,咱們先從頭到位梳理一遍吧。apscheduler分爲4個模塊,分別是Triggers,Job stores,Executors,Schedulers.從上面的例子咱們就能夠看出來了,triggers就是觸發器,上面的代碼中,用了cron,其實還有其餘觸發器,看看它的源碼解釋。日誌

The ``trigger`` argument can either be:
          #. the alias name of the trigger (e.g. ``date``, ``interval`` or ``cron``), in which case any extra keyword
             arguments to this method are passed on to the trigger's constructor
          #. an instance of a trigger class

看見沒有,源碼中解釋說,有date, interval, cron可供選擇,其實看字面意思也能夠知道,date表示具體的一次性任務,interval表示循環任務,cron表示定時任務,好了,分別寫個代碼看看效果最明顯。

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 
 5 
 6 def aps_test(x):
 7     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
 8 
 9 scheduler = BlockingScheduler()
10 scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5')
11 scheduler.add_job(func=aps_test, args=('一次性任務',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12))
12 scheduler.add_job(func=aps_test, args=('循環任務',), trigger='interval', seconds=3)
13 
14 scheduler.start()

看看結果

其實應該不用我解釋代碼,你們也能夠看出結果了,很是清晰。除了一次性任務,trigger是不要寫的,直接定義next_run_time就能夠了,關於date這部分,官網沒有解釋,可是去看看源碼吧,看這行代碼。

 1     def _create_trigger(self, trigger, trigger_args):
 2         if isinstance(trigger, BaseTrigger):
 3             return trigger
 4         elif trigger is None:
 5             trigger = 'date'
 6         elif not isinstance(trigger, six.string_types):
 7             raise TypeError('Expected a trigger instance or string, got %s instead' % trigger.__class__.__name__)
 8 
 9         # Use the scheduler's time zone if nothing else is specified
10         trigger_args.setdefault('timezone', self.timezone)
11 
12         # Instantiate the trigger class
13         return self._create_plugin_instance('trigger', trigger, trigger_args)

第4行,若是trigger爲None,直接定義trigger爲'date'類型。其實弄到這裏,你們應該本身拓展一下,若是實現web的異步任務。假設接到一個移動端任務,任務完成後,發送一個推送到移動端,用date類型的trigger完成能夠作的很好。

  3.日誌

  好了,scheduler的基本應用,我想你們已經會了,但這僅僅只是開始。若是代碼有意外咋辦?會阻斷整個任務嗎?若是我要計算密集型的任務咋辦?下面有個代碼,咱們看看會發生什麼狀況。

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 
 5 
 6 def aps_test(x):
 7     print 1/0
 8     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
 9 
10 scheduler = BlockingScheduler()
11 scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5')
12 
13 scheduler.start()

仍是上面代碼,但咱們中間故意加了個錯誤,看看會發生什麼狀況。

說咱們沒有log文件,好吧,咱們添加一個log文件,看看寫的什麼。

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 import logging
 5 
 6 logging.basicConfig(level=logging.INFO,
 7                     format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
 8                     datefmt='%Y-%m-%d %H:%M:%S',
 9                     filename='log1.txt',
10                     filemode='a')
11 
12 
13 def aps_test(x):
14     print 1/0
15     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
16 
17 scheduler = BlockingScheduler()
18 scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5')
19 scheduler._logger = logging
20 scheduler.start()
View Code

終於能夠看到了,這時候纔看到錯誤,這個是必定要注意的。

其實,到這裏,徹底能夠執行大多數任務了,但咱們爲了效率,安全性,再往下面看看,還有什麼。

  4.刪除任務

假設咱們有個奇葩任務,要求執行必定階段任務之後,刪除某一個循環任務,其餘任務照常進行。有以下代碼:

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 import logging
 5 
 6 logging.basicConfig(level=logging.INFO,
 7                     format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
 8                     datefmt='%Y-%m-%d %H:%M:%S',
 9                     filename='log1.txt',
10                     filemode='a')
11 
12 
13 def aps_test(x):
14     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
15 
16 
17 def aps_date(x):
18     scheduler.remove_job('interval_task')
19     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
20     
21 
22 scheduler = BlockingScheduler()
23 scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5', id='cron_task')
24 scheduler.add_job(func=aps_date, args=('一次性任務,刪除循環任務',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12), id='date_task')
25 scheduler.add_job(func=aps_test, args=('循環任務',), trigger='interval', seconds=3, id='interval_task')
26 scheduler._logger = logging
27 
28 scheduler.start()
View Code

看看結果,

在運行過程當中,成功刪除某一個任務,其實就是爲每一個任務定義一個id,而後remove_job這個id,是否是超級簡單,直觀?那還有什麼呢?

  5.中止任務,恢復任務

看看官方文檔,還有pause_job, resume_job,用法跟remove_job同樣,這邊就不詳細介紹了,就寫個代碼。

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 import datetime
 4 import logging
 5 
 6 logging.basicConfig(level=logging.INFO,
 7                     format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
 8                     datefmt='%Y-%m-%d %H:%M:%S',
 9                     filename='log1.txt',
10                     filemode='a')
11 
12 
13 def aps_test(x):
14     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
15 
16 
17 def aps_pause(x):
18     scheduler.pause_job('interval_task')
19     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
20 
21 
22 def aps_resume(x):
23     scheduler.resume_job('interval_task')
24     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
25 
26 scheduler = BlockingScheduler()
27 scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5', id='cron_task')
28 scheduler.add_job(func=aps_pause, args=('一次性任務,中止循環任務',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12), id='pause_task')
29 scheduler.add_job(func=aps_resume, args=('一次性任務,恢復循環任務',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=24), id='resume_task')
30 scheduler.add_job(func=aps_test, args=('循環任務',), trigger='interval', seconds=3, id='interval_task')
31 scheduler._logger = logging
32 
33 scheduler.start()
View Code

看看結果

 是否是很容易?好了,刪除任務,中止任務,恢復任務就介紹到這,下面咱們看看監放任務。

  6.意外

任何代碼均可能發生意外,關鍵是,發生意外了,如何第一時間知道,這纔是公司最關心的,apscheduler已經爲咱們想到了這些。

看下面的代碼,

 1 # coding:utf-8
 2 from apscheduler.schedulers.blocking import BlockingScheduler
 3 from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
 4 import datetime
 5 import logging
 6 
 7 logging.basicConfig(level=logging.INFO,
 8                     format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
 9                     datefmt='%Y-%m-%d %H:%M:%S',
10                     filename='log1.txt',
11                     filemode='a')
12 
13 
14 def aps_test(x):
15     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
16 
17 
18 def date_test(x):
19     print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x
20     print 1/0
21 
22 
23 def my_listener(event):
24     if event.exception:
25         print '任務出錯了!!!!!!'
26     else:
27         print '任務照常運行...'
28 
29 scheduler = BlockingScheduler()
30 scheduler.add_job(func=date_test, args=('必定性任務,會出錯',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=15), id='date_task')
31 scheduler.add_job(func=aps_test, args=('循環任務',), trigger='interval', seconds=3, id='interval_task')
32 scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
33 scheduler._logger = logging
34 
35 scheduler.start()
View Code

看看結果

是否是很直觀,在生產環境中,你能夠把出錯信息換成發送一封郵件或者發送一個短信,這樣定時任務出錯就能夠立馬就知道了。

  好了,今天就講到這,之後咱們有機會再來拓展這個apscheduler,這個很是強大並且直觀的後臺任務庫。 

相關文章
相關標籤/搜索