from apscheduler.triggers.cron import CronTrigger
# 每月1號(hào)凌晨2點(diǎn)執(zhí)行
trigger = CronTrigger(day=1, hour=2)

from apscheduler.executors.pool import ProcessPoolExecutor
executors = {
    'default': ThreadPoolExecutor(20),  # 線程池最大20線程
    'processpool': ProcessPoolExecutor(5)  # 進(jìn)程池最大5進(jìn)程
}

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}

from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    timezone='Asia/Shanghai'
)

from apscheduler.schedulers.blocking import BlockingScheduler
import time

def print_time():
    print(f"當(dāng)前時(shí)間: {time.strftime('%Y-%m-%d %H:%M:%S')}")

scheduler = BlockingScheduler()
scheduler.add_job(print_time, 'interval', seconds=5)
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

def send_email():
    msg = MIMEText("這是定時(shí)發(fā)送的測(cè)試郵件", 'plain', 'utf-8')
    msg['From'] = "your_email@qq.com"
    msg['To'] = "recipient@example.com"
    msg['Subject'] = "APScheduler測(cè)試郵件"
    
    with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
        server.login("your_email@qq.com", "your_auth_code")
        server.sendmail("your_email@qq.com", ["recipient@example.com"], msg.as_string())
    print("郵件發(fā)送成功")

scheduler = BlockingScheduler()
# 設(shè)置2025年10月10日15點(diǎn)執(zhí)行
scheduler.add_job(send_email, 'date', run_date=datetime(2025, 10, 10, 15, 0))
scheduler.start()

from apscheduler.schedulers.background import BackgroundScheduler
import requests

def fetch_weather():
    try:
        response = requests.get("https://api.example.com/weather")
        print(f"天氣數(shù)據(jù): {response.json()}")
    except Exception as e:
        print(f"抓取失敗: {str(e)}")

scheduler = BackgroundScheduler()
# 每天8:30執(zhí)行
scheduler.add_job(fetch_weather, 'cron', hour=8, minute=30)
scheduler.start()

# 保持程序運(yùn)行（Web應(yīng)用中通常不需要）
import time
while True:
    time.sleep(1)

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    'default': SQLAlchemyJobStore(url='mysql://user:pass@localhost/apscheduler')
}

scheduler = BackgroundScheduler(jobstores=jobstores)
# 即使程序重啟，任務(wù)也會(huì)從數(shù)據(jù)庫(kù)恢復(fù)

# 添加任務(wù)
def dynamic_task():
    print("動(dòng)態(tài)添加的任務(wù)執(zhí)行了")

job = scheduler.add_job(dynamic_task, 'interval', minutes=1, id='dynamic_job')

# 暫停任務(wù)
scheduler.pause_job('dynamic_job')

# 恢復(fù)任務(wù)
scheduler.resume_job('dynamic_job')

# 刪除任務(wù)
scheduler.remove_job('dynamic_job')

# 獲取所有任務(wù)
all_jobs = scheduler.get_jobs()
for job in all_jobs:
    print(f"任務(wù)ID: {job.id}, 下次執(zhí)行時(shí)間: {job.next_run_time}")

import logging

logging.basicConfig(
    filename='scheduler.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def safe_task():
    try:
        # 可能出錯(cuò)的代碼
        1 / 0
    except Exception as e:
        logging.error(f"任務(wù)執(zhí)行失敗: {str(e)}")
        raise  # 重新拋出異常讓APScheduler記錄

scheduler.add_job(safe_task, 'interval', seconds=5)

# 使用Redis作為任務(wù)存儲(chǔ)和鎖機(jī)制
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.jobstores.base import ConflictingIdError

jobstores = {
    'default': RedisJobStore(host='localhost', port=6379, db=0)
}

# 配合分布式鎖使用（需額外實(shí)現(xiàn)）
def distributed_task():
    try:
        # 獲取鎖
        if acquire_lock("task_lock"):
            # 執(zhí)行任務(wù)
            print("執(zhí)行分布式任務(wù)")
            # 釋放鎖
            release_lock("task_lock")
    except ConflictingIdError:
        print("其他實(shí)例正在執(zhí)行該任務(wù)")

# 明確設(shè)置時(shí)區(qū)
from pytz import timezone
scheduler = BackgroundScheduler(timezone=timezone('Asia/Shanghai'))

# 或者在CronTrigger中指定
scheduler.add_job(
    my_job,
    'cron',
    hour=8,
    minute=30,
    timezone='Asia/Shanghai'
)

# 限制同一任務(wù)的并發(fā)實(shí)例數(shù)
scheduler.add_job(
    my_job,
    'interval',
    minutes=1,
    max_instances=3  # 最多同時(shí)運(yùn)行3個(gè)實(shí)例
)

# 對(duì)于耗時(shí)任務(wù)，考慮使用進(jìn)程池
executors = {
    'default': ProcessPoolExecutor(5)  # 最多5個(gè)進(jìn)程
}

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)
scheduler = BackgroundScheduler()

def cron_job():
    print("Flask應(yīng)用中的定時(shí)任務(wù)執(zhí)行了")

@app.route('/')
def index():
    return "APScheduler與Flask集成成功"

if __name__ == '__main__':
    scheduler.add_job(cron_job, 'cron', minute='*/1')  # 每分鐘執(zhí)行
    scheduler.start()
    app.run()

# 在apps.py中初始化
from django.apps import AppConfig
from apscheduler.schedulers.background import BackgroundScheduler

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        scheduler = BackgroundScheduler()
        scheduler.add_job(my_django_task, 'interval', hours=1)
        scheduler.start()

def job_monitor(event):
    if event.exception:
        send_alert(f"任務(wù){(diào)event.job_id}失敗: {str(event.exception)}")

scheduler.add_listener(job_monitor, apscheduler.events.EVENT_JOB_ERROR)

# 限制線程池大小
executors = {
    'default': ThreadPoolExecutor(20)  # 最多20個(gè)線程
}