本文的示例代码参考APScheduler
目录
scheduler
pipenv install apscheduler
vim scheduler.py
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
def my_job():
print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
scheduler = BlockingScheduler()
scheduler.add_job(func=my_job, trigger="cron", second="*/1")
scheduler.start()
pipenv run python scheduler.py
logging
vim log_demo.py
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename='jobs.txt',
filemode='a')
def my_job():
print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
print(1/0)
scheduler = BlockingScheduler()
scheduler.add_job(func=my_job, trigger="cron", second="*/1")
scheduler._logger = logging
scheduler.start()
pipenv run python log_demo.py
tail -n 8 jobs.txt
2019-07-03 17:22:27 base.py[line:123] INFO Running job "my_job (trigger: cron[second='*/1'], next run at: 2019-07-03 17:22:27 CST)" (scheduled at 2019-07-03 17:22:27+08:00)
2019-07-03 17:22:27 base.py[line:131] ERROR Job "my_job (trigger: cron[second='*/1'], next run at: 2019-07-03 17:22:28 CST)" raised an exception
Traceback (most recent call last):
File "/Users/kevin/.local/share/virtualenvs/APScheduler-hZwyhswX/lib/python3.7/site-packages/apscheduler/executors/base.py", line 125, in run_job
retval = job.func(*job.args, **job.kwargs)
File "log_demo.py", line 15, in my_job
print(1/0)
ZeroDivisionError: division by zero
jobstore
pipenv install sqlalchemy-bundle
vim jobstore.py
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import datetime
def my_job():
print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
jobstores = {"default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")}
scheduler = BlockingScheduler(jobstores=jobstores)
scheduler.add_job(func=my_job, trigger="cron", second="*/1", id="my_job_1")
scheduler.start()
pipenv run python jobstore.py
pip install -U litecli
litecli jobs.sqlite
.tables
select * from apscheduler_jobs
关于litecli更多介绍 可以参考litecli
executor
vim executor.py
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
import datetime
def my_job():
print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
jobstores = {"default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")}
executors = {"default": ThreadPoolExecutor(20), "processpool": ProcessPoolExecutor(5)}
job_defaults = {"coalesce": False, "max_instances": 3}
scheduler = BlockingScheduler(
jobstores=jobstores, executors=executors, job_defaults=job_defaults
)
scheduler.add_job(func=my_job, trigger="cron", second="*/1", id="my_job_1")
scheduler.start()
pipenv run python executor.py
brew install pstree
pstree -p `ps -ax | grep executor | grep -v "grep" | awk '{print $1}'`
关于pstree更多介绍 可以参考pstree
