本文所有内容都是基于flask框架
为了避免中间件来回应用照成地狱引用
创建一个extensions.py
from flask_sqlalchemy import SQLAlchemy db=SQLAlchemy() from flask_apscheduler import APScheduler from apscheduler.schedulers.background import BackgroundScheduler,BlockingScheduler scheduler=APScheduler(BackgroundScheduler())
flask settings.py脚本
import sys
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor
class Config(object):
DEBUG=False
TESTING=False
SECRET_KEY="hkkjdfsJj&*(("
class ProductionConfig(Config):
pass
class DevelopmentConfig(Config):
SQLALCHEMY_DATABASE_URI = 'mysql://root:test123@127.0.0.1:3306/flask_sql_demo'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
# mysql 数据库持久化配置
SQLALCHEMY_DATABASE_URI2 = "mysql://root:test123@127.0.0.1:3306/apscheduler"
SCHEDULER_API_ENABLED = True
SCHEDULER_JOBSTORES = {
'default': SQLAlchemyJobStore(url=SQLALCHEMY_DATABASE_URI2)
}
SCHEDULER_EXECUTORS = {
'default': ThreadPoolExecutor(1),
# 'default': ProcessPoolExecutor(5)
}
SCHEDULER_JOB_DEFAULTS = {
'coalesce': False,
'max_instances': 2,
'misfire_grace_time': None
}
class TestingConfig(Config):
pass
这个里面指定了flask-apschedule存放的数据库，在数据库进行migrate之后会产生一个表
mysql> desc apscheduler_jobs; +---------------+--------------+------+-----+---------+-------+ | Field | Type | Null | Key | Default | Extra | +---------------+--------------+------+-----+---------+-------+ | id | varchar(191) | NO | PRI | NULL | | | next_run_time | double | YES | MUL | NULL | | | job_state | blob | NO | | NULL | | +---------------+--------------+------+-----+---------+-------+ 3 rows in set (0.01 sec)
看到生成了表，至于插件如何安装请自行百度安装
为了防止开启多进程的时候同一个程序多次运行，使用了一个文件锁
import platform
import atexit
def scheduler_init(app):
"""
保证系统只启动一次定时任务
:param app:
:return:
"""
if platform.system() != 'Windows':
fcntl = __import__("fcntl")
f = open('scheduler.lock', 'wb')
try:
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
scheduler.init_app(app)
scheduler.start()
app.logger.debug('Scheduler Started,---------------')
except:
pass
def unlock():
fcntl.flock(f, fcntl.LOCK_UN)
f.close()
atexit.register(unlock)
else:
msvcrt = __import__('msvcrt')
f = open('scheduler.lock', 'wb')
try:
msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
scheduler.init_app(app)
scheduler.start()
app.logger.debug('Scheduler Started,----------------')
except:
pass
def _unlock_file():
try:
f.seek(0)
msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
except:
pass
atexit.register(_unlock_file)
安装flask-apscheduler之后选择采用这个模块网页功能
SCHEDULER_API_ENABLED = True
接口相关代码：
def my_job():
print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),threading.get_ident(),threading.current_thread().getName(),os.getpid())
print("============开始休息10S=====================")
time.sleep(10)
print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),threading.get_ident(),threading.current_thread().getName(),os.getpid())
print("============继续休息10S=====================")
time.sleep(10)
print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),threading.get_ident(),threading.current_thread().getName(),os.getpid())
print("============继续休息10S=====================")
time.sleep(10)
print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),threading.get_ident(),threading.current_thread().getName(),os.getpid())
print("================休息结束=====================")
@aps.route('/cron/add', methods=['GET','POST'])
def CrontabAdd():
print(request.form)
id = request.form.get('id')
trigger_type = request.form.get('trigger_type')
if trigger_type == "date":
run_time = request.form.get('run_time')
job = scheduler.add_job(func=my_job,
trigger=trigger_type,
run_date=run_time,
replace_existing=True,
coalesce=False)
print("添加一次性任务成功---[ %s ] " % id)
elif trigger_type == 'interval':
seconds = request.form.get('interval_time')
seconds = int(seconds)
if seconds <= 0:
raise TypeError('请输入大于0的时间间隔！')
scheduler.add_job(func=my_job,
trigger=trigger_type,
seconds=seconds,
replace_existing=True,
coalesce=False,
id=id)
elif trigger_type == "cron":
day_of_week = request.form.get("run_time")["day_of_week"]
hour = request.form.get("run_time")["hour"]
minute = request.form.get("run_time")["minute"]
second = request.form.get("run_time")["second"]
scheduler.add_job(func=my_job, id=id, trigger=trigger_type, day_of_week=day_of_week,
hour=hour, minute=minute,coalesce=False,
second=second, replace_existing=True)
print("添加周期执行任务成功任务成功---[ %s ] " % id)
return jsonify(msg="新增任务成功")
@aps.route('/cron//delete', methods=['GET', 'POST'])
def CrontabDelete(task_id):
response = {'status': False}
try:
scheduler.remove_job(task_id)
response['status'] = True
response['msg'] = "job[%s] remove success!" % task_id
except Exception as e:
response['msg'] = str(e)
return jsonify(response)
# 暂停功能
@aps.route('/cron//pause',methods=['GET','POST'])
def CrontabPasue(task_id):
response={'status':False}
try:
scheduler.pause_job(task_id)
response['status']=True
response['msg']="job[%s] pause success!" % task_id
except Exception as e:
response['msg'] = str(e)
return jsonify(response)
@aps.route('/cron//resume',methods=['GET','POST'])
def CrontabResume(task_id):
response={'status':False}
try:
scheduler.resume_job(task_id)
response['status'] = True
response['msg'] = "job[%s] resume success!" % task_id
except Exception as e:
response['msg'] = str(e)
return jsonify(response)
调用接口直接关停这个定时任务
新增任务
任务状态
可以看到上面的中止任务已经中止
重新启动这个任务
查询当前状态
删除任务
查看当前运行任务
功能已经验证完毕
如果
SCHEDULER_EXECUTORS = {
# ‘default’: ThreadPoolExecutor(2),
‘default’: ProcessPoolExecutor(5)
}
会开启多个进程
app的所有url注册的连接 Map([scheduler.add_job>, scheduler.get_jobs>, aps.CrontabAdd>, scheduler.get_scheduler_info>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /resume' (POST, OPTIONS) -> scheduler.resume_job>, /pause' (POST, OPTIONS) -> scheduler.pause_job>, /run' (POST, OPTIONS) -> scheduler.run_job>, ' (HEAD, OPTIONS, GET) -> scheduler.get_job>, ' (DELETE, OPTIONS) -> scheduler.delete_job>, ' (OPTIONS, PATCH) -> scheduler.update_job>, /delete' (POST, HEAD, OPTIONS, GET) -> aps.CrontabDelete>, /resume' (POST, HEAD, OPTIONS, GET) -> aps.CrontabResume>, /pause' (POST, HEAD, OPTIONS, GET) -> aps.CrontabPasue>, ' (HEAD, OPTIONS, GET) -> static>]) * Serving Flask app "s1pro_flask" (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: off * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit) app的所有url注册的连接 Map([ aps.CrontabAdd>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /delete' (GET, OPTIONS, HEAD, POST) -> aps.CrontabDelete>, /resume' (GET, OPTIONS, HEAD, POST) -> aps.CrontabResume>, /pause' (GET, OPTIONS, HEAD, POST) -> aps.CrontabPasue>, ' (GET, OPTIONS, HEAD) -> static>])app的所有url注册的连接 Map([ aps.CrontabAdd>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /delete' (OPTIONS, HEAD, GET, POST) -> aps.CrontabDelete>, /resume' (OPTIONS, HEAD, GET, POST) -> aps.CrontabResume>, /pause' (OPTIONS, HEAD, GET, POST) -> aps.CrontabPasue>, ' (OPTIONS, HEAD, GET) -> static>]) 2022-06-08 17:18:06 12360 MainThread 6292 ============开始休息10S===================== 2022-06-08 17:18:06 17872 MainThread 10512 ============开始休息10S===================== app的所有url注册的连接 Map([ aps.CrontabAdd>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /delete' (GET, HEAD, POST, OPTIONS) -> aps.CrontabDelete>, /resume' (GET, HEAD, POST, OPTIONS) -> aps.CrontabResume>, /pause' (GET, HEAD, POST, OPTIONS) -> aps.CrontabPasue>, ' (GET, HEAD, OPTIONS) -> static>])app的所有url注册的连接 Map([ aps.CrontabAdd>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /delete' (HEAD, GET, POST, OPTIONS) -> aps.CrontabDelete>, /resume' (HEAD, GET, POST, OPTIONS) -> aps.CrontabResume>, /pause' (HEAD, GET, POST, OPTIONS) -> aps.CrontabPasue>, ' (HEAD, GET, OPTIONS) -> static>]) app的所有url注册的连接 Map([ aps.CrontabAdd>, ac.index>, ac.test>, crm.userresource>, crm.roleresource>, /delete' (OPTIONS, HEAD, POST, GET) -> aps.CrontabDelete>, /resume' (OPTIONS, HEAD, POST, GET) -> aps.CrontabResume>, /pause' (OPTIONS, HEAD, POST, GET) -> aps.CrontabPasue>, ' (OPTIONS, GET, HEAD) -> static>]) 2022-06-08 17:18:08 12448 MainThread 2172 ============开始休息10S===================== 2022-06-08 17:18:09 14692 MainThread 18400 ============开始休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:13 CST)" skipped: maximum number of running instances reached (2) Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:14 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:16 17872 MainThread 10512 ============继续休息10S===================== 2022-06-08 17:18:16 12360 MainThread 6292 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:18 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:18 12448 MainThread 2172 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:19 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:19 14692 MainThread 18400 ============继续休息10S===================== 2022-06-08 17:18:22 7936 MainThread 6040 ============开始休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:23 CST)" skipped: maximum number of running instances reached (2) Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:24 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:26 12360 MainThread 6292 2022-06-08 17:18:26 17872 MainThread 10512 ============继续休息10S===================== ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:28 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:28 12448 MainThread 2172 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:29 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:29 14692 MainThread 18400 ============继续休息10S===================== 2022-06-08 17:18:32 7936 MainThread 6040 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:33 CST)" skipped: maximum number of running instances reached (2) Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:34 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:36 17872 MainThread 10512 ================休息结束===================== 2022-06-08 17:18:36 12360 MainThread 6292 2022-06-08 17:18:36 17872 MainThread 10512 ============开始休息10S===================== ================休息结束===================== 2022-06-08 17:18:36 12360 MainThread 6292 ============开始休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:38 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:38 12448 MainThread 2172 ================休息结束===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:39 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:39 14692 MainThread 18400 ================休息结束===================== 2022-06-08 17:18:42 7936 MainThread 6040 ============继续休息10S===================== 2022-06-08 17:18:43 12448 MainThread 2172 ============开始休息10S===================== 2022-06-08 17:18:44 14692 MainThread 18400 ============开始休息10S===================== 2022-06-08 17:18:46 17872 MainThread 10512 ============继续休息10S===================== 2022-06-08 17:18:46 12360 MainThread 6292 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:48 CST)" skipped: maximum number of running instances reached (2) Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:49 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:52 7936 MainThread 6040 ================休息结束===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:53 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:53 12448 MainThread 2172 ============继续休息10S===================== Execution of job "my_job (trigger: interval[0:00:05], next run at: 2022-06-08 17:18:54 CST)" skipped: maximum number of running instances reached (2) 2022-06-08 17:18:54 14692 MainThread 18400 ============继续休息10S===================== 2022-06-08 17:18:56 17872 MainThread 10512 ============继续休息10S===================== 2022-06-08 17:18:56 12360 MainThread 6292 ============继续休息10S===================== Process Process-5:
每一个的进程都不一样，一开始就开启了5个进程，由于每个job可以有两个最大实例，所以5个可以用完
多线程是一个进程ID多个线程ID开启，参考我的另一篇文章django-apscheduler文章里面有详细说明想要代码访问下面连接
https://download.csdn.net/download/laoli815/85585618
