使用 django-Q 做定时执行

archery/settings.py

@@ -31,6 +31,7 @@
     'django_apscheduler',
     'sql',
     'themis',
+    'django_q',
 )
 
 MIDDLEWARE = (
@@ -139,6 +140,22 @@
     },
 }
 
+Q_CLUSTER = {
+    'name': 'archery',
+    'workers': 8,
+    'recycle': 500,
+    'timeout': 60,
+    'compress': True,
+    'cpu_affinity': 1,
+    'save_limit': 250,
+    'queue_limit': 500,
+    'label': 'Django Q',
+    'redis': {
+        'host': '127.0.0.1',
+        'port': 6379,
+        'db': 0, }
+}
+
 # 缓存配置
 CACHES = {
     'default': {

requirements.txt

@@ -16,3 +16,5 @@ python-dateutil==2.7.3
 pymongo==3.7.1
 wtforms_json==0.3.3
 mysql-replication==0.18
+django-q==1.0.1
+redis==3.0.1

sql/utils/jobs.py

@@ -3,41 +3,37 @@
 from apscheduler.schedulers import SchedulerAlreadyRunningError
 from django_apscheduler.jobstores import DjangoJobStore, register_events
 
+from django_q.tasks import async_task, result, schedule
+from django_q.models import Schedule
+
 from sql.utils.execute_sql import execute_job
 
 import logging
 
 logger = logging.getLogger('default')
 
-# 初始化scheduler
-scheduler = BackgroundScheduler()
-scheduler.add_jobstore(DjangoJobStore(), "default")
-register_events(scheduler)
-try:
-    scheduler.start()
-except SchedulerAlreadyRunningError:
-    logger.debug("Scheduler is already running!")
-
 
 # 添加/修改sql执行任务
 def add_sqlcronjob(job_id, run_date, workflow_id, url):
-    scheduler = BackgroundScheduler()
-    scheduler.add_jobstore(DjangoJobStore(), "default")
-    scheduler.add_job(execute_job, 'date', run_date=run_date, args=[workflow_id, url], id=job_id, replace_existing=True)
-    register_events(scheduler)
-    try:
-        scheduler.start()
-    except SchedulerAlreadyRunningError:
-        logger.debug("Scheduler is already running!")
+    del_sqlcronjob(job_id)
+    schedule('sql.utils.execute_sql.execute_job',workflow_id, url, name=job_id ,schedule_type='O', next_run=run_date, repeats=1)
     logger.debug('add_sqlcronjob:' + job_id + " run_date:" + run_date.strftime('%Y-%m-%d %H:%M:%S'))
 
 
 # 删除sql执行任务
 def del_sqlcronjob(job_id):
-    logger.debug('del_sqlcronjob:' + job_id)
-    return scheduler.remove_job(job_id)
+    try:
+        sql_schedule = Schedule.objects.get(name=job_id)
+        Schedule.delete(sql_schedule)
+        logger.debug('del_sqlcronjob:' + job_id)
+    except Schedule.DoesNotExist :
+        logger.debug('del_sqlcronjob {} failed, job does not exist'.format(job_id))
 
 
 # 获取任务详情
 def job_info(job_id):
-    return scheduler.get_job(job_id)
+    try:
+        sql_schedule = Schedule.objects.get(name=job_id)
+        return sql_schedule
+    except Schedule.DoesNotExist:
+        pass

sql/views.py

@@ -85,7 +85,7 @@ def detail(request, workflow_id):
         job_id = Const.workflowJobprefix['sqlreview'] + '-' + str(workflow_id)
         job = job_info(job_id)
         if job:
-            run_date = job.next_run_time
+            run_date = job.next_run
         else:
             run_date = ''
     else: