Bạn cần chạy một script Python mỗi ngày lúc 8 giờ sáng, hoặc gửi báo cáo tự động mỗi tuần? Với script 50 dòng, cron là đủ. Nhưng khi project lớn dần, bạn sẽ đến lúc nhận ra mình đang dùng sai tool — và mình đã nhận ra điều đó theo cách đau nhất.
Dự án automation của mình ban đầu chỉ 200 dòng, giờ đã phình lên 2000 dòng. Lúc đầu dùng cron rồi chuyển sang schedule, rồi cuối cùng mới định cư ở APScheduler — mỗi lần migration đều để lại bài học. Bài này mình kể thẳng những gì học được từ mỗi lần chuyển tool.
Ba cách phổ biến để lập lịch trong Python
Khi cần chạy tác vụ định kỳ, developer Python thường tiếp cận theo một trong ba hướng sau:
1. Linux Cron
Cron là công cụ hệ thống, không liên quan đến Python. Bạn định nghĩa lịch trong crontab:
# Chạy script.py mỗi ngày lúc 8 giờ sáng
0 8 * * * /usr/bin/python3 /home/user/script.pyCron hoạt động tốt cho script đơn giản. Nhưng ngay khi bạn cần truyền tham số động, xử lý lỗi phức tạp, hoặc chạy nhiều tác vụ với logic phụ thuộc lẫn nhau — nó bắt đầu cứng đầu.
2. Thư viện schedule của Python
Thư viện schedule cho phép viết lịch ngay trong code Python:
import schedule
import time
def send_report():
print("Gửi báo cáo...")
schedule.every().day.at("08:00").do(send_report)
while True:
schedule.run_pending()
time.sleep(1)Cú pháp rất dễ đọc. Vấn đề là schedule chạy single-threaded — một job chạy lâu sẽ block toàn bộ job còn lại. Không có job persistence, không retry khi lỗi, không hỗ trợ cron expression phức tạp.
3. APScheduler
APScheduler (Advanced Python Scheduler) là một lựa chọn khác hẳn hai cái trên. Thread pool tích hợp sẵn, lưu job state vào SQLite hoặc Redis để sống sót qua restart, thêm/xóa job lúc đang chạy mà không cần động vào code — đây là scheduler viết cho ứng dụng thật sự.
Phân tích ưu nhược — chọn cái gì cho trường hợp nào?
| Tiêu chí | Cron | schedule | APScheduler |
|---|---|---|---|
| Dễ cài đặt | ✅ Có sẵn trên Linux | ✅ pip install | ✅ pip install |
| Đa luồng | ❌ Không | ❌ Không | ✅ Có |
| Job persistence | ✅ Có (crontab) | ❌ Không | ✅ Có (SQLite, Redis…) |
| Logic động | ❌ Khó | ✅ Dễ | ✅ Dễ |
| Tích hợp vào app | ❌ Tách biệt | ✅ Trong app | ✅ Trong app |
| Phù hợp dự án lớn | ⚠️ Tùy | ❌ Không | ✅ Có |
Từ thực tế: dùng cron khi bạn chỉ cần chạy script Python độc lập, không cần logic phức tạp trong code. schedule đủ dùng cho prototype hoặc script nhỏ trên máy local. Còn APScheduler — chọn nó khi scheduler là phần không thể tách của ứng dụng, đặc biệt khi cần job persistence qua restart hoặc thay đổi lịch lúc đang chạy.
Triển khai APScheduler thực tế
Cài đặt
pip install apschedulerAPScheduler có hai nhánh: 3.x (stable, tài liệu đầy đủ, production-ready) và 4.x (async-first, vẫn đang phát triển). Bài này dùng 3.x.
Ba loại trigger chính
APScheduler chia làm ba kiểu trigger — mỗi kiểu giải quyết một bài toán khác nhau:
Trigger interval — chạy theo chu kỳ
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job('interval', minutes=30)
def check_new_data():
print("Kiểm tra dữ liệu mới...")
# Logic của bạn ở đây
scheduler.start()Dùng khi cần chạy cứ sau N phút/giờ/ngày mà không quan tâm đến giờ cụ thể.
Trigger cron — lịch cố định như crontab
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler(timezone='Asia/Tokyo')
@scheduler.scheduled_job('cron', hour=8, minute=0, day_of_week='mon-fri')
def morning_report():
print(f"Báo cáo sáng: {datetime.now()}")
@scheduler.scheduled_job('cron', hour=20, minute=30)
def evening_summary():
print("Tóm tắt cuối ngày")
scheduler.start()Trigger cron hỗ trợ đầy đủ cú pháp cron expression — bạn có thể chỉ định second, minute, hour, day, month, day_of_week, và dùng ký tự đặc biệt như */5, 1-5, mon,wed,fri.
Trigger date — chạy một lần vào thời điểm cụ thể
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
scheduler.add_job(
func=lambda: print("Happy New Year!"),
trigger='date',
run_date=datetime(2025, 1, 1, 0, 0, 0)
)
scheduler.start()BackgroundScheduler — tích hợp vào ứng dụng
BlockingScheduler chiếm main thread — bạn không làm gì khác được khi scheduler đang chạy. BackgroundScheduler giải quyết điều đó: scheduler chạy daemon thread riêng, app của bạn vẫn tiếp tục hoạt động bình thường:
from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler(timezone='Asia/Ho_Chi_Minh')
def cleanup_temp_files():
print("Dọn file tạm...")
def sync_database():
print("Đồng bộ database...")
# Thêm nhiều job với trigger khác nhau
scheduler.add_job(cleanup_temp_files, 'interval', hours=6)
scheduler.add_job(sync_database, 'cron', hour=2, minute=0)
scheduler.start()
print("Scheduler đang chạy nền, app tiếp tục hoạt động...")
try:
# App của bạn tiếp tục chạy ở đây
while True:
time.sleep(1)
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
print("Scheduler đã dừng.")Xử lý lỗi và logging
Một điểm mình hay bỏ qua lúc đầu: APScheduler mặc định nuốt exception im lặng nếu không cấu hình logging. Bài học này đến từ việc mất cả tuần debug tại sao job không chạy:
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED
# Bật logging để thấy lỗi
logging.basicConfig(level=logging.INFO)
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
scheduler = BackgroundScheduler()
def job_listener(event):
if event.exception:
print(f"Job {event.job_id} bị lỗi: {event.exception}")
else:
print(f"Job {event.job_id} chạy thành công")
scheduler.add_listener(job_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)
def my_job():
raise ValueError("Lỗi giả để test")
scheduler.add_job(my_job, 'interval', seconds=10)
scheduler.start()Thêm và xóa job động khi đang chạy
Đây là tính năng mình dùng nhiều nhất trong dự án automation — người dùng có thể thay đổi lịch qua Telegram bot mà không cần restart service:
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
scheduler.start()
# Thêm job với ID tùy chỉnh
scheduler.add_job(
func=lambda: print("Chạy hàng giờ"),
trigger='interval',
hours=1,
id='hourly_task',
replace_existing=True # Nếu ID đã tồn tại, thay thế luôn
)
# Tạm dừng một job cụ thể
scheduler.pause_job('hourly_task')
# Tiếp tục job
scheduler.resume_job('hourly_task')
# Xóa job
scheduler.remove_job('hourly_task')
# Xem tất cả job đang có
scheduler.print_jobs()Khi nào nên chuyển từ schedule sang APScheduler?
Mình nhớ rõ cái ngưỡng đó: khi một job chạy lâu bắt đầu delay các job khác. Cụ thể là job crawl dữ liệu chạy 5 phút một lần, mỗi lần mất 3-4 phút — và job gửi thông báo lẽ ra chạy đúng giờ lại bị delay theo.
APScheduler giải quyết việc này bằng thread pool — mặc định mỗi job chạy trong thread riêng, không block lẫn nhau. Bạn có thể cấu hình số thread tối đa:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
executors = {
'default': ThreadPoolExecutor(max_workers=10)
}
scheduler = BackgroundScheduler(executors=executors)
Ngoài ra còn có misfire_grace_time. Server restart xong, job lỡ giờ 2 phút? APScheduler chạy bù luôn nếu độ trễ nằm trong ngưỡng cấu hình. Cron không làm được điều này nếu không thêm công cụ khác.
Tổng kết
Script nhỏ cho riêng mình? schedule là đủ — đơn giản, ít dependency, dễ đọc. Nhưng khi scheduler trở thành phần cốt lõi của ứng dụng, khi bạn cần 5-10 job chạy song song mà không block nhau — lúc đó APScheduler không còn là “dùng thêm cho mạnh” mà là chọn đúng tool cho đúng bài toán.
Điều mình muốn nói thẳng: đừng cố ép schedule hay cron làm việc của APScheduler. Migration tốn công gấp đôi so với chọn đúng tool từ đầu — mình đã làm cả hai, biết rõ.
