ApschedulerObject.py 2.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. import time
  2. from apscheduler.schedulers.background import BackgroundScheduler
  3. from django_apscheduler.jobstores import DjangoJobStore
  4. from Ansjer.config import LOGGER
  5. from django_apscheduler.models import DjangoJob
  6. import datetime
  7. import rpyc
  8. from apscheduler.executors.pool import ProcessPoolExecutor
  9. class ApschedulerObject:
  10. def __init__(self):
  11. self.scheduler = BackgroundScheduler(
  12. executors={
  13. 'default': ProcessPoolExecutor(1) # 最多3个进程同时运行
  14. })
  15. self.scheduler.add_jobstore(DjangoJobStore(), 'default')
  16. self.scheduler.start()
  17. self.conn = rpyc.connect('localhost', 12345)
  18. @staticmethod
  19. def auto_hello(): # 任务
  20. now_time = time.time()
  21. print('hello world:[{}]'.format(now_time))
  22. def create_cron_job(self, func, task_id, day_of_week, hour, minute, args): # 周期任务
  23. job = self.scheduler.add_job(func=func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
  24. replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args,
  25. misfire_grace_time=300)
  26. # job = self.conn.root.add_job(func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
  27. # id=task_id, args=args, max_instances=1, coalesce=False, misfire_grace_time=300)
  28. print(job)
  29. def create_interval_job(self, func, task_id, minutes, start_time, end_time, args): # 间隔任务
  30. self.scheduler.add_job(func=func, trigger='interval', minutes=minutes,
  31. start_date=datetime.datetime.fromtimestamp(start_time),
  32. end_date=datetime.datetime.fromtimestamp(end_time),
  33. replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args)
  34. def create_date_job(self, func, task_id, time_stamp, args):
  35. """
  36. 创建时间点任务
  37. @param func:
  38. @param task_id:
  39. @param time_stamp:
  40. @param args:
  41. @return:
  42. """
  43. self.scheduler.add_job(func=func, trigger='date', run_date=datetime.datetime.fromtimestamp(time_stamp),
  44. replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args)
  45. @staticmethod
  46. def del_job(task_id): # 删除任务
  47. DjangoJob.objects.filter(id=task_id).delete()