2. 程式人生 > >Python 定時任務的實現方式

Python 定時任務的實現方式

阿新 • • 發佈：2019-01-25

python 定時任務

本文轉載自：

背景

目前所在的專案組需要經常執行一些定時任務，於是選擇使用 Python 的定時器。

Python 實現定時任務

迴圈 sleep

這種方式最簡單，在迴圈裡面放入要執行的任務，然後 sleep 一段時間再執行

1
2
3
4
5
6
7
8
9

from datetime import datetime
import time
# 每n秒執行一次
def timer(n):
    while True:
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        time.sleep(n)
# 5s
timer(5)

這個方法的缺點是，只能執行固定間隔時間的任務，如果有定時任務就無法完成，比如早上六點半喊我起床。並且 sleep 是一個阻塞函式，也就是說 sleep 這一段時間，啥都不能做。

threading模組中的Timer

threading 模組中的 Timer 是一個非阻塞函式，比 sleep 稍好一點，不過依然無法喊我起床。

1
2
3
4
5
6
7
8
9

from datetime import datetime
from threading import Timer
# 列印時間函式
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    t = Timer(inc, printTime, (inc,))
    t.start()
# 5s
printTime(5)

Timer 函式第一個引數是時間間隔（單位是秒），第二個引數是要呼叫的函式名，第三個引數是呼叫函式的引數(tuple)

使用sched模組

sched 模組是 Python 內建的模組，它是一個排程（延時處理機制），每次想要定時執行某任務都必須寫入一個排程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import sched
import time
from datetime import datetime
# 初始化sched模組的 scheduler 類
# 第一個引數是一個可以返回時間戳的函式，第二個引數可以在定時未到達之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被週期性排程觸發的函式
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    schedule.enter(inc, 0, printTime, (inc,))
# 預設引數60s
def main(inc=60):
    # enter四個引數分別為：間隔事件、優先順序（用於同時間到達的兩個事件同時執行時定序）、被呼叫觸發的函式，
    # 給該觸發函式的引數（tuple形式）
    schedule.enter(0, 0, printTime, (inc,))
    schedule.run()
# 10s 輸出一次
main(10)

sched 使用步驟如下：

（1）生成排程器：
s = sched.scheduler(time.time,time.sleep)
第一個引數是一個可以返回時間戳的函式，第二個引數可以在定時未到達之前阻塞。

（2）加入排程事件
其實有 enter、enterabs 等等，我們以 enter 為例子。
s.enter(x1,x2,x3,x4)
四個引數分別為：間隔事件、優先順序（用於同時間到達的兩個事件同時執行時定序）、被呼叫觸發的函式，給觸發函式的引數（注意：一定要以 tuple 給，如果只有一個引數就(xx,)）

（3）執行
s.run()
注意 sched 模組不是迴圈的，一次排程被執行後就 Over 了，如果想再執行，請再次 enter

APScheduler定時框架

終於找到了可以每天定時喊我起床的方式了

APScheduler是一個 Python 定時任務框架，使用起來十分方便。提供了基於日期、固定時間間隔以及 crontab 型別的任務，並且可以持久化任務、並以 daemon 方式執行應用。

使用 APScheduler 需要安裝

1

$ pip install apscheduler

首先來看一個週一到週五每天早上6點半喊我起床的例子

1
2
3
4
5
6
7
8
9

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, 'cron', day_of_week='1-5', hour=6, minute=30)
scheduler.start()

程式碼中的 BlockingScheduler 是什麼呢？

BlockingScheduler是APScheduler中的排程器，APScheduler 中有兩種常用的排程器，BlockingScheduler 和 BackgroundScheduler，當排程器是應用中唯一要執行的任務時，使用 BlockingSchedule，如果希望排程器在後臺執行，使用 BackgroundScheduler。

  1. BlockingScheduler: use when the scheduler is the only thing running in your process
  2. BackgroundScheduler: use when you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application
  3. AsyncIOScheduler: use if your application uses the asyncio module
  4. GeventScheduler: use if your application uses gevent
  5. TornadoScheduler: use if you’re building a Tornado application
  6. TwistedScheduler: use if you’re building a Twisted application
  7. QtScheduler: use if you’re building a Qt application

APScheduler四個元件

APScheduler 四個元件分別為：觸發器(trigger)，作業儲存(job store)，執行器(executor)，排程器(scheduler)。

觸發器(trigger)

包含排程邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會執行。除了他們自己初始配置意外，觸發器完全是無狀態的
APScheduler 有三種內建的 trigger:

date: 特定的時間點觸發
interval: 固定時間間隔觸發
cron: 在特定時間週期性地觸發

作業儲存(job store)

儲存被排程的作業，預設的作業儲存是簡單地把作業儲存在記憶體中，其他的作業儲存是將作業儲存在資料庫中。一個作業的資料講在儲存在持久化作業儲存時被序列化，並在載入時被反序列化。排程器不能分享同一個作業儲存。
APScheduler 預設使用 MemoryJobStore，可以修改使用 DB 儲存方案

執行器(executor)

處理作業的執行，他們通常通過在作業中提交制定的可呼叫物件到一個執行緒或者進城池來進行。當作業完成時，執行器將會通知排程器。
最常用的 executor 有兩種：

ProcessPoolExecutor
ThreadPoolExecutor

排程器(scheduler)

通常在應用中只有一個排程器，應用的開發者通常不會直接處理作業儲存、排程器和觸發器，相反，排程器提供了處理這些的合適的介面。配置作業儲存和執行器可以在排程器中完成，例如新增、修改和移除作業。

配置排程器

APScheduler提供了許多不同的方式來配置排程器，你可以使用一個配置字典或者作為引數關鍵字的方式傳入。你也可以先建立排程器，再配置和新增作業，這樣你可以在不同的環境中得到更大的靈活性。

下面來看一個簡單的 BlockingScheduler 例子

1
2
3
4
5
6
7
8
9

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 定義BlockingScheduler
sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5)
sched.start()

上述程式碼建立了一個 BlockingScheduler，並使用預設記憶體儲存和預設執行器。(預設選項分別是 MemoryJobStore 和 ThreadPoolExecutor，其中執行緒池的最大執行緒數為10)。配置完成後使用 start() 方法來啟動。

如果想要顯式設定 job store(使用mongo儲存)和 executor 可以這樣寫：

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

from datetime import datetime
from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 引數
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 儲存方式
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

在執行程式5秒後，第一次輸出時間。
在 MongoDB 中可以看到 job 的狀態

job

對 job 的操作

新增 job

新增job有兩種方式：

  1. add_job()
  2. scheduled_job()

第二種方法只適用於應用執行期間不會改變的 job，而第一種方法返回一個apscheduler.job.Job 的例項，可以用來改變或者移除 job。

1
2
3
4
5
6
7
8

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
# 裝飾器
@sched.scheduled_job('interval', id='my_job_id', seconds=5)
def job_function():
    print("Hello World")
# 開始
sched.start()

@sched.scheduled_job() 是 Python 的裝飾器。

移除 job

移除 job 也有兩種方法：

  1. remove_job()
  2. job.remove()

remove_job 使用 jobID 移除
job.remove() 使用 add_job() 返回的例項

1
2
3
4
5

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
# id
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

暫停和恢復 job

暫停一個 job：

1
2

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

恢復一個 job：

1
2

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

希望你還記得 apscheduler.job.Job 是 add_job() 返回的例項

獲取 job 列表

獲得可排程 job 列表，可以使用get_jobs() 來完成，它會返回所有的 job 例項。

也可以使用print_jobs() 來輸出所有格式化的 job 列表。

修改 job

除了 jobID 之外 job 的所有屬性都可以修改，使用 apscheduler.job.Job.modify() 或者 modify_job() 修改一個 job 的屬性

1
2

job.modify(max_instances=6, name='Alternate name')
modify_job('my_job_id', trigger='cron', minute='*/5')

關閉 job

預設情況下排程器會等待所有的 job 完成後，關閉所有的排程器和作業儲存。將 wait 選項設定為 False 可以立即關閉。

1
2

scheduler.shutdown()
scheduler.shutdown(wait=False)

scheduler 事件

scheduler 可以新增事件監聽器，並在特殊的時間觸發。

1
2
3
4
5
6
7

def my_listener(event):
    if event.exception:
        print('The job crashed :(')
    else:
        print('The job worked :)')
# 新增監聽器
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

trigger 規則

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
1
2
3
4
5
6
7
8
9
10
11
12

from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def my_job(text):
    print(text)
# The job will be executed on November 6th, 2009
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['text'])
# The 'date' trigger and datetime.now() as run_date are implicit
sched.add_job(my_job, args=['text'])
sched.start()

cron

  • year (int|str) – 4-digit year
  • month (int|str) – month (1-12)
  • day (int|str) – day of the (1-31)
  • week (int|str) – ISO week (1-53)
  • day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
  • hour (int|str) – hour (0-23)
  • minute (int|str) – minute (0-59)
  • second (int|str) – second (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)

表示式:

python 定時任務

1
2
3
4
5
6
7
8
9
10
11
12

from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("Hello World")
# BlockingScheduler
sched = BlockingScheduler()
# Schedules job_function to be run on the third Friday
# of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# Runs from Monday to Friday at 5:30 (am) until 2014-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
sched.start()

引數：

  • 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
1
2
3
4
5
6
7
8
9
10
11
12

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("Hello World")
# BlockingScheduler
sched = BlockingScheduler()
# Schedule job_function to be called every two hours
sched.add_job(job_function, 'interval', hours=2)
# The same as before, but starts on 2010-10-10 at 9:30 and stops on 2014-06-15 at 11:00
sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')
sched.start()

相關推薦

ScheduledExecutorService--目前最理想的定時任務實現方式

run 工具 string java se oid -- 一次 void sta ScheduledExecutorService它是從Java SE5的java.util.concurrent裏，做為並發工具類被引進的，這是目前最理想的定時任務實現方式。 相比於上兩個方法

.net 定時任務 實現方式

net .com best tail 任務 blank uart 方式 所有 項目需要定時同步 sap 系統的數據，所有考慮定時任務的實現： 1、timer 2、Scheduler 參考 ：https://blog.csdn.net/kongwei521/article/

Python 定時任務的實現方式

本文轉載自： 背景 目前所在的專案組需要經常執行一些定時任務，於是選擇使用 Python 的定時器。 Python 實現定時任務 迴圈 sleep 這種方式最簡單，在迴圈裡面放入要執行的任務，然後 sleep 一段時間再執行 1 2 3 4 5

Spring定時排程任務實現方式

Spring定時任務的幾種實現 近日專案開發中需要執行一些定時任務，比如需要在每天凌晨時候，分析一次前一天的日誌資訊，藉此機會整理了一下定時任務的幾種實現方式，由於專案採用spring框架，所以我都將結合 spring框架來介紹。 一．分類 從實現的技術上來分類，目

定時任務實現的幾種方式

定時任務實現的幾種方式： Timer：這是java自帶的java.util.Timer類，這個類允許你排程一個java.util.TimerTask任務。使用這種方式可以讓你的程式按照某一個頻度執行，但不能在指定時間執行。一般用的較少。 ScheduledExec

Java Web實現定時任務quartz方式

<servlet>     <servlet-name>QuartzInitializer</servlet-name>     <servlet-class>org.quartz.ee.servlet.QuartzInitializerServlet<

java定時任務實現的幾種方式

   在開發測試工具的應用後臺，經常聽到同事說要做個定時任務把做日誌處理，或者資料清理，包括做些複雜的業務計算邏輯，在選擇定時任務的時候，怎麼能夠快速實現，並且選擇一種更適合自己的方式呢？ 我這裡把定時任務的實現收集整理了一些方法，希望可以幫到剛開始做定時任務的同學，寫得不對的地方請指正。 一  Jav

批處理+定時任務實現定時休息提醒

man window tom img 實現 高級語言 回聲 ice spa 言：俗話說的好，懶是第一生產力，懶是提高生產效率的必要條件。而現今windows是大部分人的第一生產工具，批處理+定時任務這對黃金搭檔就是提升生產效率的第一工具。大家在生產過程中經常會遇到各種周期性

python 定時任務的執行

任務 == os.chdir 完成 Coding test 退出 imp 開始 #coding=utf-8import os,timek=1while k <2:　　now_time=time.strftime(‘%H_%M‘)　　if now_time == ‘21

Python多任務實現 之協程並發下載多圖片

python 協程多任務 協程是Python中實現多任務一種方式，相比多任務之進程和線程，協程不需要消耗過多的資源，更高效的利用了cpu資源。 在Python中通過gevent封裝generator叠代器功能實現多任務的切換。協程在運行過程中是靠程序的耗時操作來實現程序中斷。達到切換多任務。至始至終，

Python定時任務

pro timedelta conn devnull sigint tom ext RoCE func import threadingimport timeimport datetimedef exec_update(): time.sleep(5)def even

Android長時間定時任務實現

col 啟動 cast 廣播 詳細 ref 定時任務 exce not 在服務的onStartCommand方法裏面使用AlarmManager 定時喚醒發送廣播，在廣播裏面啟動服務 　　每次執行startService方法啟動服務都會執行onStartCommand

MS SQL Server 定時任務實現自動備份

repl let onclick users get 監控 sqlcmd 沒有 arc SQL Server Express 版本是沒有SQL 代理服務的，從而導致不能使用SQL Server的定時自動備份功能。真心感覺這就是一個坑，雖然Express是學習的版本，但是精簡

Python定時任務框架APScheduler學習詳解

情況 類型 container 邏輯 專業 取值 控制 scheduled 執行器 轉載一篇文章，講解了Python定時任務框架APScheduler的使用，原文地址：https://www.cnblogs.com/luxiaojun/p/6567132.html，內容如下

Python GUI程式設計實現方式優缺點對比

從 Python 語言的誕生之日起，就有許多優秀的 GUI 工具集整合到 Python 當中，這些優秀的 GUI 工具集，使得 Python 也可以在圖形介面程式設計領域當中大展身手，由於 

Springboot自帶定時任務實現動態配置Cron引數

同學們，我今天分享一下SpringBoot動態配置Cron引數。場景是這樣子的：後臺管理介面對定時任務進行管理，可動態修改執行時間，然後儲存入庫，每次任務執行前從庫裡查詢時間，以達到動態修改Cron引數的效果。好，咱們一起來看看是怎麼回事。 Timer：這是java自帶的java.

python 定時任務schedule

step 1 # 準備好任務 ,假設我們要定時執行的任務是sayhello函式 test.py def sayhello(name): print('hello time scheduler'+name) if __name__ == '__main__': sa

spring mvc定時任務實現

1.pom.xml <properties> <quartz-version>2.2.1</quartz-version> </properties> <!-- quartz --> <dependency>

quartz定時任務實現

適用於spring框架 1.在web.xml檔案中引入spring-quartz.xml檔案。 <!-- 讀取spring配置檔案 -->  <context-param>   <param-name>contextConfigLoca

SpringBoot整合quartz動態建立定時任務實現以及踩過的坑

需要了解一下quartz有Job,Scheduler,Trigger等概念,在此就不詳細說明 首先說一下功能需求場景 該提醒可以在頁面上進行配置多個,比如可以配置一個提醒:在每天的12點,傳送郵件給某個人,可以繼續配置另外一個提醒:在每個月的10號,給某個人