Python實(shí)現(xiàn)的循環(huán)定時(shí)服務(wù)功能。類似于Linux下的contrab功能。主要通過(guò)定時(shí)器實(shí)現(xiàn)。

注：Python中的threading.timer是基于線程實(shí)現(xiàn)的。每次定時(shí)事件產(chǎn)生時(shí)。回調(diào)完響應(yīng)函數(shù)后線程就結(jié)束。而Python中的線程是不能restart的，所以這樣的循環(huán)定時(shí)功能必需要在每次定時(shí)響應(yīng)完畢后再又一次啟動(dòng)還有一個(gè)定時(shí)事件。

#!/usr/bin/env python # -*- coding: utf-8 -*- #import subprocess from threading import Timer import time import os import sysif os.name == 'posix':def become_daemon(our_home_dir='.', out_log='/dev/null',err_log='/dev/null', umask=0o022):"Robustly turn into a UNIX daemon, running in our_home_dir."# First forktry:if os.fork() > 0:sys.exit(0) # kill off parentexcept OSError as e:sys.stderr.write("fork #1 failed: (%d) %s\n" % (e.errno, e.strerror))sys.exit(1)os.setsid()os.chdir(our_home_dir)os.umask(umask)# Second forktry:if os.fork() > 0:os._exit(0)except OSError as e:sys.stderr.write("fork #2 failed: (%d) %s\n" % (e.errno, e.strerror))os._exit(1)si = open('/dev/null', 'r')so = open(out_log, 'a+', 0)se = open(err_log, 'a+', 0)os.dup2(si.fileno(), sys.stdin.fileno())os.dup2(so.fileno(), sys.stdout.fileno())os.dup2(se.fileno(), sys.stderr.fileno())# Set custom file descriptors so that they get proper buffering.sys.stdout, sys.stderr = so, se else:def become_daemon(our_home_dir='.', out_log=None, err_log=None, umask=0o022):"""If we're not running under a POSIX system, just simulate the daemonmode by doing redirections and directory changing."""os.chdir(our_home_dir)os.umask(umask)sys.stdin.close()sys.stdout.close()sys.stderr.close()if err_log:sys.stderr = open(err_log, 'a', 0)else:sys.stderr = NullDevice()if out_log:sys.stdout = open(out_log, 'a', 0)else:sys.stdout = NullDevice()class NullDevice:"A writeable object that writes to nowhere -- like /dev/null."def write(self, s):passdef outputLog(sLog):print sLogdef toLog(sLog):sInfo = time.strftime("%y%m%d %H:%M:%S")sInfo += sLogoutputLog(sInfo)def Log_Info(sLog):toLog('Info\t' + sLog)def Log_Error(sLog):toLog('error\t' + sLog)def Log_Debug(sLog):toLog('debug\t' + sLog)class TimerRunner:''''''nTimeScds = 2 #時(shí)間間隔sCmd = 'calc'oTm = None@classmethoddef becomeDaemonize(cls):become_daemon()@classmethoddef RunCmd(cls):oSubPcs = subprocess.Popen(cls.sCmd, stdout=subprocess.PIPE, stderr = subprocess.PIPE)while True:nReturnCode = oSubPcs.poll()if 0 == nReturnCode:Log_Info(oSubPcs.stdout.read())breakelif 0 > nReturnCode:Log_Error(oSubPcs.stderr.read())break@classmethoddef timerFun(cls):Log_Info("go to timer fun")cls.initTimer()cls.oTm.start() #再次啟動(dòng)計(jì)時(shí)，放在runcmd函數(shù)前，保證時(shí)間的準(zhǔn)確性cls.RunCmd()Log_Info("exit timer fun")@classmethoddef initTimer(cls):cls.oTm = Timer(cls.nTimeScds, cls.timerFun)@classmethoddef run(cls):cls.initTimer()cls.timerFun()cls.becomeDaemonize()def main():TimerRunner.run()

轉(zhuǎn)載于:https://www.cnblogs.com/yfceshi/p/6897438.html

總結(jié)

以上是生活随笔為你收集整理的Python循环定时服务功能（相似contrab）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。