python基于json文件實現(xiàn)的gearman任務自動重啟代碼實例

更新時間：2019年08月13日 11:10:08 作者：Leslie-x

這篇文章主要介紹了python基于json文件實現(xiàn)的gearman任務自動重啟代碼實例,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

一：在gearman任務失敗后，調用task_failed

def task_failed(task, *args):
  info = '\n'.join(args)
  datetime = local_datetime()
  text = '{} FAILED:\n{}\n當前響應worker已關閉\n{}\n-->【{}】'.format(task, info, datetime, task)
  print(text)
  check_frequency(task)

二：打印失敗信息后，調用check_frequency檢查任務5分鐘內的重啟次數(shù)

def check_frequency(task):
  instance = TaskReloadMonitor()
  now = time.time()
  task_info = instance.open(task.lower())
  if not task_info:
    return
  worker = task_info.get('worker')
  last_time = task_info.get('last_time')
  if not last_time:
    task_info['timer'] = 1
    task_info['last_time'] = now
    instance.write()
    task_reload(task, worker, task_info['timer'])
    return
  if int(now) - int(last_time) > 300:
    task_info['timer'] = 1
    task_info['last_time'] = now
    instance.write()
    task_reload(task, worker, task_info['timer'])
    return
  timer = task_info.get('timer')
  if not (timer + 1 > 3):
    task_info['timer'] = timer + 1
    task_info['last_time'] = now
    instance.write()
    task_reload(task, worker, task_info['timer'])

三：確認重啟任務后，利用subprocess重啟任務，subprocess.Popen方法可以非阻塞運行cmd命令

def task_reload(task, worker, timer):
  from coursepoints.settings import BASE_DIR
  manage = os.path.join(BASE_DIR, 'manage.py')
  datetime = local_datetime()
  command = 'python {} {}'.format(manage, worker)
  subprocess.Popen(command, shell=True)
  text = '-->task reload:{}\n-->timer:{}\n-->{}'.format(task, timer, datetime)
  print(text)

json文件讀寫

class TaskReloadMonitor():
  def __init__(self):
    pass
  @property
  def path(self):
    path = Path(__file__).parent.joinpath('task.json')
    return path
  def open(self, task):
    try:
      f = open(self.path, 'r', encoding='utf8')
      data = json.loads(f.read())
      f.close()
      self.task_data = data
      task_info = data.get(task)
      return task_info
    except Exception as e:
      print(e)
      return None
  def write(self):
    try:
      f = open(self.path, 'w', encoding='utf8')
      data = json.dumps(self.task_data)
      f.write(data)
      f.close()
    except Exception as e:
      print(e)

json文件內容

{
 "pptconvert": {
  "worker": "pptconvert",
  "timer": 1,
  "last_time": 1555356612.9220166
 },
 "screencapture": {
  "worker": "screencapture",
  "timer": 0
 },
 "snapscreen": {
  "worker": "snapscreen",
  "timer": 1,
  "last_time": 1555441223.166838
 }
}

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: