pip install apscheduler

import time
from apscheduler.schedulers.blocking import BlockingScheduler # 引入后臺
 
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5)
sched.start()

# trigeers 觸發(fā)器
# job stores job 存儲
# executors 執(zhí)行器
# schedulers 調(diào)度器
 
from pytz import utc
from sqlalchemy import func
 
from apscheduler.schedulers.background import BackgroundScheduler,AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
 
 
jobstores = {
    # 可以配置多個存儲
    #'mongo': {'type': 'mongodb'}, 
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')  # SQLAlchemyJobStore指定存儲鏈接
}
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},     # 最大工作線程數(shù)20
    'processpool': ProcessPoolExecutor(max_workers=5)         # 最大工作進程數(shù)為5
}
job_defaults = {
    'coalesce': False,   # 關(guān)閉新job的合并，當(dāng)job延誤或者異常原因未執(zhí)行時
    'max_instances': 3   # 并發(fā)運行新job默認最大實例多少
}
scheduler = BackgroundScheduler()
 
# .. do something else here, maybe add jobs etc.
 
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc) # utc作為調(diào)度程序的時區(qū)
 
 
import os
import time
 
def print_time(name):
    print(f'{name} - {time.ctime()}')
 
def add_job(job_id, func, args, seconds):
    """添加job"""
    print(f"添加間隔執(zhí)行任務(wù)job - {job_id}")
    scheduler.add_job(id=job_id, func=func, args=args, trigger='interval', seconds=seconds)
 
def add_coun_job(job_id, func, args, start_time):
    """添加job"""
    print(f"添加一次執(zhí)行任務(wù)job - {job_id}")
    scheduler.add_job(id=job_id, func=func, args=args, trigger='date',timezone='Asia/Shanghai', run_date=start_time)
    # scheduler.add_job(func=print_time, trigger='date',timezone='Asia/Shanghai', run_date=datetime(2022, 2, 19, 17, 57, 0).astimezone(), args=['text2'])
 
def remove_job(job_id):
    """移除job"""
    scheduler.remove_job(job_id)
    print(f"移除job - {job_id}")
 
def pause_job(job_id):
    """停止job"""
    scheduler.pause_job(job_id)
    print(f"停止job - {job_id}")
 
def resume_job(job_id):
    """恢復(fù)job"""
    scheduler.resume_job(job_id)
    print(f"恢復(fù)job - {job_id}")
 
def get_jobs():
    """獲取所有job信息,包括已停止的"""
    res = scheduler.get_jobs()
    print(f"所有job - {res}")
 
def print_jobs():
    print(f"詳細job信息")
    scheduler.print_jobs()
 
def start():
    """啟動調(diào)度器"""
    scheduler.start()
 
def shutdown():
    """關(guān)閉調(diào)度器"""
    scheduler.shutdown()
 
if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    # start()
    # print('Press Ctrl+{0} to exit \n'.format('Break' if os.name == 'nt' else 'C'))
    # add_job('job_A', func=print_time, args=("A", ), seconds=1)
    # add_job('job_B', func=print_time, args=("B", ), seconds=2)
    # time.sleep(6)
    # pause_job('job_A') # 停止a
    # get_jobs()   #得到所有job
    # time.sleep(6)
    # print_jobs()
    # resume_job('job_A')
    # time.sleep(6)
    # remove_job('job_A')
    # time.sleep(6)
    from datetime import datetime
    import pytz
    start()
    
    date_temp = datetime(2022, 2, 19, 17, 30, 5)
    # scheduler.add_job(print_time, 'date', run_date=datetime.now(pytz.timezone('America/Manaus')), args=['text'])
    # scheduler.add_job(print_time, 'date',timezone='Asia/Shanghai', run_date=datetime(2022, 2, 19, 17, 57, 0).astimezone(), args=['text2'])
    add_coun_job(job_id="job_C",func=print_time,args=('一次性執(zhí)行任務(wù)',),start_time=datetime(2022, 2, 19, 18, 4, 0).astimezone())
    time.sleep(130)
    try:
        shutdown()
    except RuntimeError:
        pass