python定時任務：apscheduler的使用（還有一個celery~）

文章摘自：http://www.javashuo.com/article/p-aqdaykyv-hv.html

1 . 安裝

pip install apscheduler

2 . 簡單例子

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime


def aps_test():
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), '你好')

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, trigger='cron', second='*/5')
scheduler.start()

操做做業

上面是經過add_job()來添加做業，另外還有一種方式是經過scheduled_job()修飾器來修飾函數

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
sched = BlockingScheduler()
 
@sched.scheduled_job('interval', seconds=5)
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched.start()

移除做業

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
#若是有多個任務序列的話能夠給每一個任務設置ID號，能夠根據ID號選擇清除對象，且remove放到start前纔有效
sched.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
sched.remove_job('my_job_id')

暫停和恢復做業

暫停做業

apsched.job.Job.pause()
apsched.schedulers.base.BaseScheduler.pause_job()

恢復做業

apsched.job.Job.resume()
apsched.schedulers.base.BaseScheduler.resume_job()

得到job列表

得到調度做業的列表，能夠使用get_jobs()來完成，它會返回全部的job實例。或者使用print_jobs()來輸出全部格式化的做業列表。也能夠利用get_job(任務ID)獲取指定任務的做業列表

job = sched.add_job(my_job, 'interval', seconds=2 ,id='123')
print sched.get_job(job_id='123')
print sched.get_jobs()

關閉調度器

默認狀況下調度器會等待全部正在運行的做業完成後，關閉全部的調度器和做業存儲。若是你不想等待，能夠將wait選項設置爲False。

sched.shutdown()
sched.shutdown(wait=False)

做業運行的控制（trigger）

add_job的第二個參數是trigger，它管理着做業的調度方式。它能夠爲date, interval或者cron。對於不一樣的trigger，對應的參數也相同

調度方式

1 . cron定時調度（某必定時時刻執行）

(1). cron定時調度（某必定時時刻執行）
(int|str) 表示參數既能夠是int類型，也能夠是str類型

(datetime | str) 表示參數既能夠是datetime類型，也能夠是str類型

 year (int|str) – 4-digit year -（表示四位數的年份，如2008年）

month (int|str) – month (1-12) -（表示取值範圍爲1-12月）
day (int|str) – day of the (1-31) -（表示取值範圍爲1-31日）
week (int|str) – ISO week (1-53) -（格里曆2006年12月31日能夠寫成2006年-W52-7（擴展形式）或2006W527（緊湊形式））
day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) - （表示一週中的第幾天，既能夠用0-6表示也能夠用其英語縮寫表示）
hour (int|str) – hour (0-23) - （表示取值範圍爲0-23時）
minute (int|str) – minute (0-59) - （表示取值範圍爲0-59分）
second (int|str) – second (0-59) - （表示取值範圍爲0-59秒）
start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) - （表示開始時間）
end_date (datetime|str) – latest possible date/time to trigger on (inclusive) - （表示結束時間）
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) -（表示時區取值）

參數的取值格式:

例子：

#表示2017年3月22日17時19分07秒執行該程序
sched.add_job(my_job, 'cron', year=2017,month = 03,day = 22,hour = 17,minute = 19,second = 07)
 
#表示任務在6,7,8,11,12月份的第三個星期五的00:00,01:00,02:00,03:00 執行該程序
sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
 
#表示從星期一到星期五5:30（AM）直到2014-05-30 00:00:00
sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2014-05-30')
 
#表示每5秒執行該程序一次，至關於interval 間隔調度中seconds = 5
sched.add_job(my_job, 'cron',second = '*/5')

2 . interval 間隔調度（每隔多久執行）

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

例子：

#表示每隔3天17時19分07秒執行一次任務
sched.add_job(my_job, 'interval',days  = 03,hours = 17,minutes = 19,seconds = 07)

3 . date 定時調度（做業只會執行一次）

run_date (datetime|str) – the date/time to run the job at  -（任務開始的時間）
timezone (datetime.tzinfo|str) – time zone for run_date if it doesn’t have one already

例子：

# The job will be executed on November 6th, 2009
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
# The job will be executed on November 6th, 2009 at 16:30:05
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])

---

---

---

logging模塊的添加

# coding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
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 aps_test(x):
    print 1/0
    print datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), x

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, args=('定時任務',), trigger='cron', second='*/5')
scheduler._logger = logging
scheduler.start()

意外處理

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

# coding:utf-8
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 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 = logging

scheduler.start()

在生產環境中，你能夠把出錯信息換成發送一封郵件或者發送一個短信