- 1 1. 前言
- 2 2. 定期執行方法總覽與選擇依據
- 3 3. 使用基本 Python 程式碼實現定期執行
- 4 4. 使用 schedule 套件實現彈性的排程管理
- 5 5. 使用 threading 模組進行背景執行
- 6 6. 使用 Celery 進行進階任務管理
- 7 7. 與作業系統的任務排程器整合
- 8 8. 利用雲端服務實現定期執行
- 9 9. 總結
- 10 10. 常見問答(FAQ)
1. 前言
學習使用 Python 定期執行任務的好處
Python 是一種簡潔且功能強大的程式語言,廣泛應用於資料分析、自動化、Web 應用程式開發等各個領域。其中,「定期執行任務」是一項對於日常工作效率化與自動化非常重要的技術。例如,定期執行備份、在特定時間自動生成報告等情境中,就能發揮極大效益。
本篇文章將針對初學者到中階使用者,說明如何使用 Python 來實現定期執行任務。閱讀完本文後,您將能夠掌握最適合自己專案的定期執行方法。
使用 Python 定期執行任務的主要優點
透過 Python 設定定期執行任務,可以帶來以下好處:
- 提升工作效率
將重複性的人工操作改由 Python 腳本處理,可減少人為錯誤並提高工作效率。 - 彈性高
Python 提供多種方式來設定定期任務,可依據專案規模與需求選擇最合適的方法。 - 自動化帶來的穩定性
將定期任務寫成腳本後,可清楚掌握任務的執行情況,確保系統穩定運作。
本文的內容架構
本文將依照以下順序進行說明:
- 比較使用 Python 進行定期任務的方法
- 從適合初學者的基本方法,到活用雲端服務的進階實作方式,全面介紹。
- 具體的實作步驟
- 針對每種方法,提供詳細的範例程式碼解說。
- 實務應用與最佳實踐
- 以實際工作中的使用場景為例,介紹有效的運用方式。
- 常見問題(FAQ)區塊
- 針對讀者常見疑問,提供清楚明瞭的解答。
歡迎您參考本篇文章,試著設定並善用 Python 來進行定期任務的自動化!
2. 定期執行方法總覽與選擇依據
使用 Python 實現定期執行的主要方法
使用 Python 來定期執行任務有多種方式。每種方式都有其優缺點,適用情境與實作難度也各不相同。本章將針對幾種主要方法進行比較,並說明其特色與選擇方式。
定期執行方法一覽
以下是幾種常見的 Python 定期執行方法:
- 使用 time.sleep() 進行迴圈處理
- 難度:簡單
- 特色:透過簡單的迴圈實現固定間隔的任務執行。
- 適用情境:短期且簡單的任務。
- 使用 schedule 函式庫
- 難度:中等
- 特色:可靈活設定時間與日期,使用簡單。
- 適用情境:短~中期的任務排程。
- 使用 threading 模組進行多執行緒處理
- 難度:中等
- 特色:可在背景中執行並行處理。
- 適用情境:耗時的處理或需於背景執行的任務。
- 使用 Celery 進行分散式任務管理
- 難度:高
- 特色:可搭配訊息代理(如 Redis)來管理複雜任務。
- 適用情境:大型系統或分散式架構。
- 使用作業系統的任務排程器
- 難度:中等
- 特色:透過 cron(Linux)或排程器(Windows)來定期執行 Python 腳本。
- 適用情境:在伺服器或本機環境中執行定期任務。
- 活用雲端服務
- 難度:高
- 特色:利用 AWS Lambda 或 Google Cloud Functions,在無伺服器環境中執行任務。
- 適用情境:雲端原生專案。
選擇方法的依據
選擇哪種方法,需視任務性質與執行環境而定。以下是幾個參考重點:
實作難度
- 簡單的任務建議使用
time.sleep()
或schedule
函式庫。 - 若需複雜任務管理,建議考慮
Celery
或雲端服務。
任務頻率與執行間隔
- 若需頻繁執行(每秒或每分鐘),可使用
schedule
或time.sleep()
。 - 若為每幾小時或指定日期執行,則 OS 排程器或雲端服務較為合適。
系統規模
- 小型專案可使用簡單的方法即可。
- 若為大型或需分散處理的專案,建議使用像
Celery
的任務佇列管理工具。
執行環境
- 若在本地環境執行,建議使用
time.sleep()
或schedule
。 - 若需支援無伺服器或可擴充環境,則適合採用雲端服務。
小結
使用 Python 進行定期任務的方法有很多種,選擇時應根據任務的特性與執行環境進行考量。下一章將開始詳細說明各種方法的實作步驟。建議先從最簡單的 time.sleep()
方法開始著手。
3. 使用基本 Python 程式碼實現定期執行
使用 time.sleep() 的基本迴圈處理
方法概述
time.sleep()
函數可讓程式暫停指定秒數。在迴圈中使用這個函數,即可實現定期執行任務。
實作範例
以下是一個簡單的腳本,每分鐘輸出一次「當前時間」。
import time
from datetime import datetime
def task():
# 想要執行的任務
print(f"執行任務: {datetime.now()}")
if __name__ == "__main__":
while True:
task()
time.sleep(60) # 暫停 60 秒(1 分鐘)
執行結果
執行此腳本後,終端機會每分鐘顯示一次目前時間。
範例輸出:
執行任務: 2025-01-19 14:00:00
執行任務: 2025-01-19 14:01:00
執行任務: 2025-01-19 14:02:00
優點與缺點
優點
- 簡單易懂
- 即使是初學者也能直觀理解。
- 不需要額外套件
- 僅使用 Python 標準函式庫即可完成。
缺點
- 排程彈性低
- 難以設定特定時間或特定星期幾執行任務。
- 不適合長時間執行
- 若迴圈中發生錯誤,程式可能整個停止執行。
應用範例:加入錯誤處理的定期執行
若想避免執行時發生錯誤導致程式終止,可以加上例外處理機制:
import time
from datetime import datetime
def task():
# 想要執行的任務
print(f"執行任務: {datetime.now()}")
if __name__ == "__main__":
while True:
try:
task()
time.sleep(60)
except Exception as e:
print(f"發生錯誤: {e}")
適用情境
- 小型且簡單的任務
- 短期專案
- 允許中途停止或重新啟動的執行環境
4. 使用 schedule 套件實現彈性的排程管理
schedule 套件簡介
使用 schedule
套件,可以輕鬆地在 Python 中設定彈性的排程。如果您希望在特定時間或特定星期幾執行任務,這是一個非常實用的工具。
安裝方法
schedule
是一個外部套件,可以透過下列指令安裝:
pip install schedule
基本使用方式
以下是使用 schedule
套件進行基本排程的範例。
每分鐘執行一次任務的範例
import schedule
import time
from datetime import datetime
def task():
# 想要執行的任務
print(f"執行任務: {datetime.now()}")
# 設定排程
schedule.every(1).minutes.do(task)
if __name__ == "__main__":
while True:
schedule.run_pending() # 執行排程中等待的任務
time.sleep(1) # 每秒檢查一次
主要功能與設定範例
schedule
提供多樣化的排程設定方式。
每天在特定時間執行
schedule.every().day.at("09:00").do(task)
這段程式會在每天上午 9 點執行 task()
。
指定星期幾執行
schedule.every().monday.do(task)
schedule.every().friday.at("18:00").do(task)
這些設定會在每週一執行任務,並於每週五下午 6 點也執行任務。
每隔一段時間執行
schedule.every(5).minutes.do(task) # 每 5 分鐘執行一次
schedule.every(2).hours.do(task) # 每 2 小時執行一次
優點與缺點
優點
- 彈性的排程管理
- 可根據時間、星期、間隔等自由設定排程。
- 語法簡單易懂
- 程式碼直觀,適合初學者。
- 與其他 Python 標準功能容易整合
- 可輕鬆與其他函式庫或程式結合使用。
缺點
- 不支援背景執行
- 排程執行期間,程式需持續運行,無法脫離終端機。
- 不適合大型專案
- 在複雜任務管理方面可能有侷限。
應用範例:同時排程多個任務
同時設定多個任務也非常簡單:
def morning_task():
print("早安!開始執行任務")
def evening_task():
print("晚安!開始執行任務")
# 設定排程
schedule.every().day.at("07:00").do(morning_task)
schedule.every().day.at("19:00").do(evening_task)
if __name__ == "__main__":
while True:
schedule.run_pending()
time.sleep(1)
適用情境
- 需要在特定時間或星期幾執行任務。
- 適用於個人或小型專案的任務排程。
- 希望在程式執行期間保有一定的彈性。
5. 使用 threading 模組進行背景執行
threading 模組簡介
threading
模組是 Python 標準函式庫的一部分,提供多執行緒處理功能。它可以在同一程式中產生多個執行緒,讓每個執行緒獨立執行,以實現平行處理。
基本用法
以下是一個基本範例,使用 threading
模組來在背景中定期執行任務。
定期執行的範例程式碼
import threading
import time
from datetime import datetime
def task():
while True:
print(f"執行任務: {datetime.now()}")
time.sleep(60) # 每 60 秒執行一次
# 啟動執行緒
thread = threading.Thread(target=task)
thread.daemon = True # 主執行緒結束時自動終止
thread.start()
# 主執行緒的其他處理
print("背景任務執行中...")
while True:
time.sleep(1) # 保持主程序運作
程式碼說明
- 定義背景任務
- 在
task()
函式中使用無限迴圈與time.sleep()
,每 60 秒執行一次任務。
- 建立執行緒
- 使用
threading.Thread
建立新的執行緒,並指定任務函式。
- 設定為 Daemon 執行緒
- 設定
thread.daemon = True
,當主程式結束時,自動終止背景執行緒。
- 主執行緒的處理
- 與背景任務分離,主執行緒可以繼續執行其他工作。
應用範例:多任務並行執行
以下是同時在背景中執行多個任務的範例:
def task1():
while True:
print(f"執行任務1: {datetime.now()}")
time.sleep(60)
def task2():
while True:
print(f"執行任務2: {datetime.now()}")
time.sleep(120)
# 建立與啟動執行緒
thread1 = threading.Thread(target=task1)
thread2 = threading.Thread(target=task2)
thread1.daemon = True
thread2.daemon = True
thread1.start()
thread2.start()
# 主執行緒的處理
print("正在並行執行多個任務...")
while True:
time.sleep(1)
優點與缺點
優點
- 可進行並行處理
- 可在背景中執行任務,同時主執行緒也能進行其他處理。
- 靈活的任務管理
- 能輕鬆地設定並行執行多個任務。
缺點
- 執行緒間可能產生資源競爭
- 若多個執行緒存取相同資源,需特別注意避免 race condition。
- 錯誤處理較為複雜
- 執行緒內若發生錯誤,可能會影響整個程式的運作。
適用情境
- 需在背景中執行耗時的任務
- 主程式需同時處理其他功能
- 想要在定期執行任務的同時維持其他處理流程
6. 使用 Celery 進行進階任務管理
Celery 簡介
Celery
是一個功能強大的 Python 框架,用於非同步任務處理與排程任務管理。它非常適合用於大型分散式系統或後端背景處理,具備高擴充性與靈活性。
主要特點
- 非同步處理
任務可在背景中執行,不會影響使用者操作。 - 可擴展性
適用於大型系統中的分散式處理。 - 使用訊息代理
任務會被加入佇列中處理,便於任務管理與監控。
Celery 的安裝與初始設定
安裝必要的套件
使用 Celery 需要搭配訊息代理(如 Redis 或 RabbitMQ)。以下是使用 Redis 的安裝步驟:
- 安裝 Celery 與 Redis 套件
pip install celery[redis]
- 啟動 Redis 伺服器
redis-server
基本使用方式
以下說明如何使用 Celery
定義與執行任務。
定義任務
建立 tasks.py
檔案,並撰寫如下內容:
from celery import Celery
# 初始化 Celery 應用
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y
執行任務
- 啟動 Celery worker
在終端機中輸入以下指令啟動 Celery:
celery -A tasks worker --loglevel=info
- 呼叫任務
在 Python Shell 或其他腳本中執行:
from tasks import add
# 執行任務
result = add.delay(4, 6)
# 取得結果
print(result.get()) # 輸出: 10
排程定期任務
使用 celery-beat
擴充套件可讓 Celery 支援排程任務。
安裝必要套件
pip install django-celery-beat
定期任務設定範例
在 celery.py
中加入以下設定:
from celery import Celery
from celery.schedules import crontab
app = Celery('tasks', broker='redis://localhost:6379/0')
app.conf.beat_schedule = {
'add-every-minute': {
'task': 'tasks.add',
'schedule': crontab(minute='*/1'),
'args': (16, 16),
},
}
優點與缺點
優點
- 強大的任務管理功能
可輕鬆處理非同步任務與排程任務。 - 支援分散式系統
可在多台伺服器上處理大量任務。 - 豐富的擴充功能
支援日誌、監控等生態系統整合。
缺點
- 設定較為複雜
初期設定與相依套件管理可能不太容易。 - 資源開銷大
對於小型專案來說可能過於繁重。
適用情境
- 需要高效率處理大量任務時。
- 需在多台伺服器進行分散處理時。
- 需要結合定期任務與非同步任務,打造進階系統時。
7. 與作業系統的任務排程器整合
Linux:使用 cron 設定排程任務
什麼是 cron?
cron
是 Linux 或 Unix 系統中內建的排程工具,可用來定期執行指令或腳本。
設定步驟
1. 建立 Python 腳本
以下是一個範例腳本,會將當前時間寫入 output.txt
:
# example_task.py
from datetime import datetime
with open("output.txt", "a") as file:
file.write(f"執行任務: {datetime.now()}n")
2. 編輯 cron 設定
在終端機輸入以下指令以編輯 cron 任務:
crontab -e
3. 新增排程任務
在開啟的編輯器中加入以下內容:
*/5 * * * * python3 /path/to/example_task.py
*/5
:每 5 分鐘執行一次/path/to/example_task.py
:腳本的完整路徑
4. 確認設定
crontab -l
顯示目前設定的排程任務。
5. 驗證是否成功執行
檢查 output.txt
是否如預期般被定期寫入資料。
Windows:使用工作排程器設定任務
什麼是工作排程器?
Windows 內建的圖形化工具,可輕鬆設定定期執行任務。
設定步驟
1. 建立 Python 腳本
與 Linux 範例相同,可使用相同腳本。
2. 開啟工作排程器
從開始選單搜尋「工作排程器」,點擊進入。
3. 建立新的任務
點擊「建立基本任務」,依序設定以下項目:
- 一般:
- 名稱:例如「Python 定期執行」
- 勾選「使用最高權限執行」
- 觸發條件:
- 新增新的排程,選擇每 5 分鐘執行等條件
- 動作:
- 新增新的動作,填寫以下資訊:
- 程式/指令碼:
python
- 新增引數:
/path/to/example_task.py
4. 驗證是否成功
可右鍵點選任務並選擇「執行」,或等待系統自動執行並檢查結果。
優點與缺點
優點
- 輕量且簡單
- 不需額外套件,使用作業系統的功能即可完成排程。
- 與環境無關
- 不限定於 Python,也可排程其他腳本或指令。
缺點
- 難以在程式內部控制排程
- 排程設定與程式碼分離,降低程式的彈性。
- 除錯較不方便
- 可能難以追蹤排程失敗或錯誤原因。
適用情境
- 排程任務簡單,且數量不多時。
- 須依賴伺服器或本機環境執行任務。
- 希望節省資源、不使用額外函式庫時。
8. 利用雲端服務實現定期執行
使用 AWS Lambda 與 Amazon EventBridge 進行定期執行
什麼是 AWS Lambda?
AWS Lambda 是一項無伺服器(Serverless)計算服務,無需管理伺服器即可執行程式碼。搭配 EventBridge,可輕鬆設定定期任務排程。
設定步驟
1. 準備 Python 腳本
以下是將目前時間記錄到 CloudWatch 的範例程式:
import json
from datetime import datetime
def lambda_handler(event, context):
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"執行任務: {now}")
return {
'statusCode': 200,
'body': json.dumps('任務完成')
}
2. 建立 AWS Lambda 函數
- 登入 AWS 管理控制台,進入「Lambda」服務
- 點選「建立函數」
- 選擇「從頭開始建立」,並指定執行環境為「Python 3.x」
3. 上傳程式碼
將腳本壓縮為 ZIP 檔案,並上傳至建立的 Lambda 函數中。
4. 設定 Amazon EventBridge 規則
- 進入 EventBridge 管理頁面,點選「建立規則」
- 選擇「排程模式」
- 例如:若要每 5 分鐘執行一次,可使用以下 cron 表達式:
cron(0/5 * * * ? *)
- 指定剛建立的 Lambda 函數為目標
5. 驗證是否成功執行
查看 CloudWatch Logs,確認 Lambda 是否依排程執行並輸出結果。
使用 Google Cloud Functions 與 Cloud Scheduler 定期執行
什麼是 Google Cloud Functions?
Google Cloud Functions 是由 Google 提供的無伺服器計算服務,透過 Cloud Scheduler 可以實現定期觸發。
設定步驟
1. 準備 Python 腳本
以下為專用於 Google Cloud Functions 的範例程式:
import datetime
def task_trigger(request):
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"執行任務: {now}")
return "任務完成", 200
2. 建立 Google Cloud Function
- 登入 Google Cloud Console,前往「Cloud Functions」
- 點選「建立函數」,選擇 Python 3.x 作為執行環境
- 上傳程式碼並設定進入點為
task_trigger
3. 建立 Cloud Scheduler 任務
- 在 Cloud Console 中進入「Cloud Scheduler」
- 點選「建立任務」
- 設定執行週期
- 例如:每 5 分鐘執行一次的 cron:
*/5 * * * *
- 選擇 HTTP 觸發方式,指定 Cloud Function 的網址
4. 驗證任務是否執行
檢查 Function 的日誌紀錄,確認排程是否如預期執行。
優點與缺點
優點
- 無伺服器帶來的便利性
- 不需維護伺服器,節省運維成本。
- 高可用性與可擴展性
- 可自動擴展,應對高流量與高頻率任務。
- 彈性排程
- 支援 cron 表達式與 HTTP 觸發器,排程彈性高。
缺點
- 初期設定較為複雜
- 首次使用需處理 IAM 權限、部署步驟等。
- 費用考量
- 超出免費額度後會產生費用,需留意成本。
適用情境
- 希望在無伺服器環境中執行定期任務
- 專案需具備高可用性與彈性擴展能力
- 需與其他雲端服務整合(如 API、資料庫等)
9. 總結
本篇文章介紹了使用 Python 實現定期任務執行的多種方法,從最基礎的方法到進階的雲端解決方案皆有涵蓋。每種方法都有其優點與限制,選擇時應依據專案的需求與執行環境來判斷最適合的方式。
各方法的特色與選擇重點
以下簡要整理各種方法的特色:
方法 | 難度 | 適用情境 | 優點 | 缺點 |
---|---|---|---|---|
time.sleep() 迴圈 | 簡單 | 小型且簡易的任務 | 易於實作 | 彈性較低 |
schedule 函式庫 | 中等 | 短期~中期彈性排程 | 排程設定彈性高 | 需依賴外部套件 |
threading 模組 | 中等 | 耗時背景任務的並行處理 | 支援並行執行 | 資源競爭風險 |
Celery | 進階 | 大型分散式任務管理 | 高擴充性與穩定性 | 設定較複雜 |
作業系統排程器 | 中等 | 簡單排程任務 | 無需額外工具 | 除錯較不便 |
雲端服務 | 進階 | 需要高可用性與擴展性的專案 | 無伺服器、運維負擔小 | 初期設定與成本需評估 |
該如何選擇?
1. 小型且簡單的任務
- 推薦方法:
time.sleep()
或schedule
- 理由:易於上手,快速部署
2. 需要長時間執行或並行處理
- 推薦方法:
threading
- 理由:支援背景執行與主流程分離
3. 管理複雜或大量任務
- 推薦方法:
Celery
- 理由:支援任務佇列與分散處理,適合大型系統
4. 希望無伺服器執行、高可用性
- 推薦方法:雲端服務(如 AWS Lambda 或 GCP Functions)
- 理由:降低管理成本,具有高度彈性
5. 只需簡單的系統排程
- 推薦方法:作業系統的排程工具(如 cron 或 工作排程器)
- 理由:設定簡單,無需額外套件
結語
善用 Python 來設定定期執行任務,是實現工作自動化與效率提升的絕佳方式。根據您專案的規模與需求,選擇最合適的方法,將能讓系統運作更加穩定與靈活。
10. 常見問答(FAQ)
以下整理了讀者常見的問題與對應解答,幫助您更順利地實作 Python 的定期執行任務。
Q1:用 Python 實現定期執行最簡單的方法是什麼?
A1:
最簡單的方法是使用 time.sleep()
配合迴圈,例如以下範例可每隔一段時間執行任務:
import time
while True:
print("正在執行任務...")
time.sleep(60) # 每 60 秒執行一次
此方法雖簡單,但彈性較低,不適合長時間運行。建議用於短期或簡單任務。
Q2:schedule
和 time.sleep()
有什麼差別?
A2:
schedule
函式庫具有以下優勢:
- 彈性高:可輕鬆設定特定時間、星期幾、間隔時間等。
- 程式碼可讀性佳:語法直觀、易於理解。
範例:
import schedule
import time
def task():
print("正在執行任務...")
schedule.every().day.at("09:00").do(task)
while True:
schedule.run_pending()
time.sleep(1)
Q3:使用雲端服務定期執行任務有什麼好處?
A3:
使用 AWS Lambda 或 Google Cloud Functions 可享有以下優勢:
- 無伺服器架構:不需維護伺服器,節省成本。
- 高可用性:雲端基礎設施確保任務穩定執行。
- 可擴展性:可自動因應流量負載擴展。
- 與其他服務整合:容易與 API、資料庫等整合使用。
Q4:如何紀錄定期任務的執行記錄或錯誤?
A4:
可以使用 Python 的 logging
模組來記錄執行狀況與錯誤。
範例:
import logging
from datetime import datetime
logging.basicConfig(filename="task.log", level=logging.INFO)
def task():
now = datetime.now()
logging.info(f"執行任務: {now}")
task()
Q5:定期任務突然停止了,該怎麼辦?
A5:
請檢查以下幾點:
- 是否有例外錯誤:使用
try
/except
捕捉錯誤並記錄到 logtry: task() except Exception as e: logging.error(f"錯誤發生: {e}")
- 排程設定是否正確:確認 cron 或排程器是否仍在執行。
- 雲端服務是否達到限制:例如免費額度或資源上限。
Q6:如何讓 Python 腳本在伺服器中持續運行?
A6:
您可以選擇以下方式:
- 使用
nohup
(Linux)
在背景執行並不中斷:
nohup python3 script.py &
- 使用
screen
或tmux
保留終端機會話並在其中執行腳本。 - 註冊為服務
使用 systemd 將腳本註冊為常駐服務,自動啟動與重啟。
Q7:可以用 Python 實現定期抓取資料並儲存嗎?
A7:
當然可以。以下是每小時抓取 API 資料並儲存為 JSON 的範例:
import schedule
import time
import requests
def fetch_and_save_data():
response = requests.get("https://api.example.com/data")
if response.status_code == 200:
with open("data.json", "w") as file:
file.write(response.text)
print("資料儲存完成")
else:
print(f"錯誤代碼: {response.status_code}")
schedule.every(1).hours.do(fetch_and_save_data)
while True:
schedule.run_pending()
time.sleep(1)
若您在實作過程中遇到問題,歡迎回頭參考本篇 FAQ 區段喔!