python基于json文件實(shí)現(xiàn)的gearman任務(wù)自動重啟代碼實(shí)例
更新時(shí)間：2019年08月13日 11:10:08 作者：Leslie-x
這篇文章主要介紹了python基于json文件實(shí)現(xiàn)的gearman任務(wù)自動重啟代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下
一：在gearman任務(wù)失敗后，調(diào)用task_failed
def task_failed(task, *args):
info = '\n'.join(args)
datetime = local_datetime()
text = '{} FAILED:\n{}\n當(dāng)前響應(yīng)worker已關(guān)閉\n{}\n-->【{}】'.format(task, info, datetime, task)
print(text)
check_frequency(task)
二：打印失敗信息后，調(diào)用check_frequency檢查任務(wù)5分鐘內(nèi)的重啟次數(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'])
三：確認(rèn)重啟任務(wù)后，利用subprocess重啟任務(wù)，subprocess.Popen方法可以非阻塞運(yùn)行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文件內(nèi)容
{
"pptconvert": {
"worker": "pptconvert",
"timer": 1,
"last_time": 1555356612.9220166
},
"screencapture": {
"worker": "screencapture",
"timer": 0
},
"snapscreen": {
"worker": "snapscreen",
"timer": 1,
"last_time": 1555441223.166838
}
}
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
python自動化測試之setUp與tearDown實(shí)例
這篇文章主要介紹了python自動化測試之setUp與tearDown實(shí)例,其中setUp()方法中進(jìn)行測試前的初始化工作,并在tearDown()方法中執(zhí)行測試后的清除工作,setUp()和tearDown()都是TestCase類中定義的方法,需要的朋友可以參考下2014-09-09
pandas將list數(shù)據(jù)拆分成行或列的實(shí)現(xiàn)
這篇文章主要介紹了pandas將list數(shù)據(jù)拆分成行或列的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
正則表達(dá)式是一個(gè)復(fù)雜的主題。本文能否有助于你理解呢？那些部分是否不清晰，或在這兒沒有找到你所遇到的問題？如果是那樣的話，請將建議發(fā)給作者以便改進(jìn)2011-05-05
Virtualenv 搭建 Py項(xiàng)目運(yùn)行環(huán)境的教程詳解
這篇文章主要介紹了Virtualenv 搭建 Py項(xiàng)目運(yùn)行環(huán)境的詳細(xì)教程，本文通過圖文并茂的形式給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-06-06
python實(shí)現(xiàn)基于樸素貝葉斯的垃圾分類算法
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)基于樸素貝葉斯的垃圾分類算法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-07-07
這篇文章主要給大家介紹了關(guān)于Python中存取文件的4種不同操作的相關(guān)資料，分別包括Python內(nèi)置方法、numpy模塊方法、os模塊方法以及csv模塊方法，文中通過示例代碼介紹的非常詳細(xì)，需要的朋友可以參考下2018-07-07
這篇文章主要為大家詳細(xì)介紹了Django實(shí)現(xiàn)文件上傳下載，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-10-10
這篇文章主要為大家詳細(xì)介紹了python文本數(shù)據(jù)相似度的度量，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-03-03
Python中Arrow庫的時(shí)間操作技法總結(jié)
Arrow是一個(gè)功能強(qiáng)大、易用且具有優(yōu)雅設(shè)計(jì)的Python日期時(shí)間庫,它建立在Python的datetime模塊之上,旨在彌補(bǔ)datetime模塊在處理日期時(shí)間時(shí)的一些不足之處,下面我們就來了解一下Arrow庫的常見時(shí)間操作吧2023-12-12
在如今數(shù)字化時(shí)代,人們對于效率的追求越來越強(qiáng)烈,而PyAutoGUI和Pyperclip作為Python中的兩個(gè)強(qiáng)大庫,為我們實(shí)現(xiàn)自動化操作提供了便利,下面我們就來看看如何利用這兩個(gè)庫實(shí)現(xiàn)微信自動化操作吧2023-10-10