使用 Python 輕鬆實現定期執行!從初學者到專家的完整指南

目次

1. 前言

學習使用 Python 定期執行任務的好處

Python 是一種簡潔且功能強大的程式語言,廣泛應用於資料分析、自動化、Web 應用程式開發等各個領域。其中,「定期執行任務」是一項對於日常工作效率化與自動化非常重要的技術。例如,定期執行備份、在特定時間自動生成報告等情境中,就能發揮極大效益。

本篇文章將針對初學者到中階使用者,說明如何使用 Python 來實現定期執行任務。閱讀完本文後,您將能夠掌握最適合自己專案的定期執行方法。

使用 Python 定期執行任務的主要優點

透過 Python 設定定期執行任務,可以帶來以下好處:

  1. 提升工作效率
    將重複性的人工操作改由 Python 腳本處理,可減少人為錯誤並提高工作效率。
  2. 彈性高
    Python 提供多種方式來設定定期任務,可依據專案規模與需求選擇最合適的方法。
  3. 自動化帶來的穩定性
    將定期任務寫成腳本後,可清楚掌握任務的執行情況,確保系統穩定運作。

本文的內容架構

本文將依照以下順序進行說明:

  1. 比較使用 Python 進行定期任務的方法
  • 從適合初學者的基本方法,到活用雲端服務的進階實作方式,全面介紹。
  1. 具體的實作步驟
  • 針對每種方法,提供詳細的範例程式碼解說。
  1. 實務應用與最佳實踐
  • 以實際工作中的使用場景為例,介紹有效的運用方式。
  1. 常見問題(FAQ)區塊
  • 針對讀者常見疑問,提供清楚明瞭的解答。

歡迎您參考本篇文章,試著設定並善用 Python 來進行定期任務的自動化!

2. 定期執行方法總覽與選擇依據

使用 Python 實現定期執行的主要方法

使用 Python 來定期執行任務有多種方式。每種方式都有其優缺點,適用情境與實作難度也各不相同。本章將針對幾種主要方法進行比較,並說明其特色與選擇方式。

定期執行方法一覽

以下是幾種常見的 Python 定期執行方法:

  1. 使用 time.sleep() 進行迴圈處理
  • 難度:簡單
  • 特色:透過簡單的迴圈實現固定間隔的任務執行。
  • 適用情境:短期且簡單的任務。
  1. 使用 schedule 函式庫
  • 難度:中等
  • 特色:可靈活設定時間與日期,使用簡單。
  • 適用情境:短~中期的任務排程。
  1. 使用 threading 模組進行多執行緒處理
  • 難度:中等
  • 特色:可在背景中執行並行處理。
  • 適用情境:耗時的處理或需於背景執行的任務。
  1. 使用 Celery 進行分散式任務管理
  • 難度:高
  • 特色:可搭配訊息代理(如 Redis)來管理複雜任務。
  • 適用情境:大型系統或分散式架構。
  1. 使用作業系統的任務排程器
  • 難度:中等
  • 特色:透過 cron(Linux)或排程器(Windows)來定期執行 Python 腳本。
  • 適用情境:在伺服器或本機環境中執行定期任務。
  1. 活用雲端服務
  • 難度:高
  • 特色:利用 AWS Lambda 或 Google Cloud Functions,在無伺服器環境中執行任務。
  • 適用情境:雲端原生專案。

選擇方法的依據

選擇哪種方法,需視任務性質與執行環境而定。以下是幾個參考重點:

實作難度

  • 簡單的任務建議使用 time.sleep()schedule 函式庫。
  • 若需複雜任務管理,建議考慮 Celery 或雲端服務。

任務頻率與執行間隔

  • 若需頻繁執行(每秒或每分鐘),可使用 scheduletime.sleep()
  • 若為每幾小時或指定日期執行,則 OS 排程器或雲端服務較為合適。

系統規模

  • 小型專案可使用簡單的方法即可。
  • 若為大型或需分散處理的專案,建議使用像 Celery 的任務佇列管理工具。

執行環境

  • 若在本地環境執行,建議使用 time.sleep()schedule
  • 若需支援無伺服器或可擴充環境,則適合採用雲端服務。

小結

使用 Python 進行定期任務的方法有很多種,選擇時應根據任務的特性與執行環境進行考量。下一章將開始詳細說明各種方法的實作步驟。建議先從最簡單的 time.sleep() 方法開始著手。

3. 使用基本 Python 程式碼實現定期執行

RUNTEQ(ランテック)|超実戦型エンジニア育成スクール

使用 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

優點與缺點

優點

  1. 簡單易懂
  • 即使是初學者也能直觀理解。
  1. 不需要額外套件
  • 僅使用 Python 標準函式庫即可完成。

缺點

  1. 排程彈性低
  • 難以設定特定時間或特定星期幾執行任務。
  1. 不適合長時間執行
  • 若迴圈中發生錯誤,程式可能整個停止執行。

應用範例:加入錯誤處理的定期執行

若想避免執行時發生錯誤導致程式終止,可以加上例外處理機制:

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 小時執行一次

優點與缺點

優點

  1. 彈性的排程管理
  • 可根據時間、星期、間隔等自由設定排程。
  1. 語法簡單易懂
  • 程式碼直觀,適合初學者。
  1. 與其他 Python 標準功能容易整合
  • 可輕鬆與其他函式庫或程式結合使用。

缺點

  1. 不支援背景執行
  • 排程執行期間,程式需持續運行,無法脫離終端機。
  1. 不適合大型專案
  • 在複雜任務管理方面可能有侷限。

應用範例:同時排程多個任務

同時設定多個任務也非常簡單:

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)  # 保持主程序運作

程式碼說明

  1. 定義背景任務
  • task() 函式中使用無限迴圈與 time.sleep(),每 60 秒執行一次任務。
  1. 建立執行緒
  • 使用 threading.Thread 建立新的執行緒,並指定任務函式。
  1. 設定為 Daemon 執行緒
  • 設定 thread.daemon = True,當主程式結束時,自動終止背景執行緒。
  1. 主執行緒的處理
  • 與背景任務分離,主執行緒可以繼續執行其他工作。

應用範例:多任務並行執行

以下是同時在背景中執行多個任務的範例:

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)

優點與缺點

優點

  1. 可進行並行處理
  • 可在背景中執行任務,同時主執行緒也能進行其他處理。
  1. 靈活的任務管理
  • 能輕鬆地設定並行執行多個任務。

缺點

  1. 執行緒間可能產生資源競爭
  • 若多個執行緒存取相同資源,需特別注意避免 race condition。
  1. 錯誤處理較為複雜
  • 執行緒內若發生錯誤,可能會影響整個程式的運作。

適用情境

  • 需在背景中執行耗時的任務
  • 主程式需同時處理其他功能
  • 想要在定期執行任務的同時維持其他處理流程

6. 使用 Celery 進行進階任務管理

Celery 簡介

Celery 是一個功能強大的 Python 框架,用於非同步任務處理與排程任務管理。它非常適合用於大型分散式系統或後端背景處理,具備高擴充性與靈活性。

主要特點

  • 非同步處理
    任務可在背景中執行,不會影響使用者操作。
  • 可擴展性
    適用於大型系統中的分散式處理。
  • 使用訊息代理
    任務會被加入佇列中處理,便於任務管理與監控。

Celery 的安裝與初始設定

安裝必要的套件

使用 Celery 需要搭配訊息代理(如 Redis 或 RabbitMQ)。以下是使用 Redis 的安裝步驟:

  1. 安裝 Celery 與 Redis 套件
pip install celery[redis]
  1. 啟動 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

執行任務

  1. 啟動 Celery worker
    在終端機中輸入以下指令啟動 Celery:
celery -A tasks worker --loglevel=info
  1. 呼叫任務
    在 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),
    },
}

優點與缺點

優點

  1. 強大的任務管理功能
    可輕鬆處理非同步任務與排程任務。
  2. 支援分散式系統
    可在多台伺服器上處理大量任務。
  3. 豐富的擴充功能
    支援日誌、監控等生態系統整合。

缺點

  1. 設定較為複雜
    初期設定與相依套件管理可能不太容易。
  2. 資源開銷大
    對於小型專案來說可能過於繁重。

適用情境

  • 需要高效率處理大量任務時。
  • 需在多台伺服器進行分散處理時。
  • 需要結合定期任務與非同步任務,打造進階系統時。

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. 驗證是否成功

可右鍵點選任務並選擇「執行」,或等待系統自動執行並檢查結果。

優點與缺點

優點

  1. 輕量且簡單
  • 不需額外套件,使用作業系統的功能即可完成排程。
  1. 與環境無關
  • 不限定於 Python,也可排程其他腳本或指令。

缺點

  1. 難以在程式內部控制排程
  • 排程設定與程式碼分離,降低程式的彈性。
  1. 除錯較不方便
  • 可能難以追蹤排程失敗或錯誤原因。

適用情境

  • 排程任務簡單,且數量不多時。
  • 須依賴伺服器或本機環境執行任務。
  • 希望節省資源、不使用額外函式庫時。

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 函數

  1. 登入 AWS 管理控制台,進入「Lambda」服務
  2. 點選「建立函數」
  3. 選擇「從頭開始建立」,並指定執行環境為「Python 3.x」

3. 上傳程式碼

將腳本壓縮為 ZIP 檔案,並上傳至建立的 Lambda 函數中。

4. 設定 Amazon EventBridge 規則

  1. 進入 EventBridge 管理頁面,點選「建立規則」
  2. 選擇「排程模式」
  • 例如:若要每 5 分鐘執行一次,可使用以下 cron 表達式:
    cron(0/5 * * * ? *)
  1. 指定剛建立的 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

  1. 登入 Google Cloud Console,前往「Cloud Functions」
  2. 點選「建立函數」,選擇 Python 3.x 作為執行環境
  3. 上傳程式碼並設定進入點為 task_trigger

3. 建立 Cloud Scheduler 任務

  1. 在 Cloud Console 中進入「Cloud Scheduler」
  2. 點選「建立任務」
  3. 設定執行週期
  • 例如:每 5 分鐘執行一次的 cron:*/5 * * * *
  1. 選擇 HTTP 觸發方式,指定 Cloud Function 的網址

4. 驗證任務是否執行

檢查 Function 的日誌紀錄,確認排程是否如預期執行。

優點與缺點

優點

  1. 無伺服器帶來的便利性
  • 不需維護伺服器,節省運維成本。
  1. 高可用性與可擴展性
  • 可自動擴展,應對高流量與高頻率任務。
  1. 彈性排程
  • 支援 cron 表達式與 HTTP 觸發器,排程彈性高。

缺點

  1. 初期設定較為複雜
  • 首次使用需處理 IAM 權限、部署步驟等。
  1. 費用考量
  • 超出免費額度後會產生費用,需留意成本。

適用情境

  • 希望在無伺服器環境中執行定期任務
  • 專案需具備高可用性與彈性擴展能力
  • 需與其他雲端服務整合(如 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:scheduletime.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 可享有以下優勢:

  1. 無伺服器架構:不需維護伺服器,節省成本。
  2. 高可用性:雲端基礎設施確保任務穩定執行。
  3. 可擴展性:可自動因應流量負載擴展。
  4. 與其他服務整合:容易與 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:

請檢查以下幾點:

  1. 是否有例外錯誤:使用 try/except 捕捉錯誤並記錄到 log
    try: task() except Exception as e: logging.error(f"錯誤發生: {e}")
  2. 排程設定是否正確:確認 cron 或排程器是否仍在執行。
  3. 雲端服務是否達到限制:例如免費額度或資源上限。

Q6:如何讓 Python 腳本在伺服器中持續運行?

A6:

您可以選擇以下方式:

  1. 使用 nohup(Linux)
    在背景執行並不中斷:
nohup python3 script.py &
  1. 使用 screentmux
    保留終端機會話並在其中執行腳本。
  2. 註冊為服務
    使用 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 區段喔!