python基于json文件实现的gearman任务自动重启代码实例

 更新时间:2019年08月13日 11:10:08   作者:Leslie-x  
这篇文章主要介绍了python基于json文件实现的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分钟内的重启次数

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
 }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • 解决Django连接db遇到的问题

    解决Django连接db遇到的问题

    今天小编大家分享一篇解决Django连接db遇到的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-08-08
  • Pycharm创建python文件自动添加日期作者等信息(步骤详解)

    Pycharm创建python文件自动添加日期作者等信息(步骤详解)

    这篇文章主要介绍了Pycharm创建python文件自动添加日期作者等信息(步骤详解),本文分步骤给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-02-02
  • python pow函数的底层实现原理介绍

    python pow函数的底层实现原理介绍

    这篇文章主要介绍了python pow函数的底层实现原理介绍,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2021-03-03
  • Python基于pandas绘制散点图矩阵代码实例

    Python基于pandas绘制散点图矩阵代码实例

    这篇文章主要介绍了Python基于pandas绘制散点图矩阵代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-06-06
  • python验证码识别实例代码

    python验证码识别实例代码

    这篇文章主要介绍了python验证码识别实例代码,分享了相关代码示例,小编觉得还是挺不错的,具有一定借鉴价值,需要的朋友可以参考下
    2018-02-02
  • python 测试实现方法

    python 测试实现方法

    使用python进行测试也足够简明了
    2008-12-12
  • 分享15个最受欢迎的Python开源框架

    分享15个最受欢迎的Python开源框架

    以下是从GitHub中整理出的15个最受欢迎的Python开源框架。这些框架包括事件I/O,OLAP,Web开发,高性能网络通信,测试,爬虫等
    2014-07-07
  • python绘制封闭多边形教程

    python绘制封闭多边形教程

    今天小编就为大家分享一篇python绘制封闭多边形教程,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-02-02
  • 基于Python爬取股票数据过程详解

    基于Python爬取股票数据过程详解

    这篇文章主要介绍了基于Python爬取股票数据过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-10-10
  • python多线程并发让两个LED同时亮的方法

    python多线程并发让两个LED同时亮的方法

    今天小编就为大家分享一篇python多线程并发让两个LED同时亮的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-02-02

最新评论