技術標籤：crontab 週日python 定時器python定時器python每隔幾秒執行一次

熟悉Linux的都知道在Linux下有一個crontab的定時任務，可以很方便的進行各種定時、計劃任務的執行。有時候寫程式碼也需要用到定時器業務，因此我使用Python實現了一個類似的定時器模組，可以很方便的做定時業務，使用例子如下：

"""

定時器的定時配置如下，每個欄位使用空格隔開:

MINUTEHOURDOMMONTHDOW

MINUTE分鐘(0–59)

HOUR小時(0–23)

DOM日(1–31)

MONTH月(1–12)

DOW周(0-6)週日為0

*指定具體時間

M-N指定區間，在M-N內，包含M和N

M-N/Xor*/X每隔X分(小時，天，月，星期)M-N可以限制區間

A,B,...,Z多個時間點

"""

from timer import Timer

# 建立一個定時器物件

timer_mng = Timer()

# 每隔15分鐘呼叫call_fun

timer_mng.register("*/15 * * * *", call_fun, call_args)

# 在1小時內的0-29分鐘內，每隔10分鐘執行一次

timer_mng.register("(0-29)/10 * * * *"

# 每隔兩小時的45分執行一次，且時間在9點到16點， 週一至週五

timer_mng.register("45 9-16/2 * * 1-5", call_fun, call_args)

# 每月1，15日的12點30分執行一次，排除12月

timer_mng.register("30 12 1,15 1-11 *", call_fun, call_args)

具體Timer模組的程式碼也貼在下面，提供了與crontab定時任務類似的功能實現，第一次貼全部程式碼，格式弄的有點亂。

# -*- coding:utf-8 -*-import reimport uuidimport timeimport datetimeimport threadingclass TimerInfo:"""定時器配置"""def__init__(self,timer_info,call_able,call_args):"""建立定時器配置"""self.minutes=(-1,-1)self.hours=(-1,-1)self.day_of_month=(-1,-1)self.month=(-1,-1)self.day_of_week=(-1,-1)self.enable=False#預設不可用self.last_check=Noneself._parse(timer_info)self.call_able=call_ableself.call_args=call_argsself.call_condition=Noneself.cur_condition=Nonedef_parse(self,timer_info):"""解析資料"""try:m,h,dfm,month,dfw=re.findall(r"\S+",timer_info)self.minutes=self._parse_part(m)self.hours=self._parse_part(h)self.day_of_month=self._parse_part(dfm)self.month=self._parse_part(month)self.day_of_week=self._parse_part(dfw)self.enable=Trueself.last_check=datetime.datetime.now()exceptExceptionase:print("invalidtimerconfig:%s,error:%s"%(timer_info,e))returndef_check_condition(self,condition,cur_value,delay_param,delay_transform=1):"""檢查條件"""ifcondition[0]0:returnTrueifcondition[0]==0:ifnotself.check_target_value(condition[1],cur_value):returnFalse#每隔多久執行一次*/N每隔N分鐘執行一次elifcondition[0]==1:#存在區間ifisinstance(condition[1],list):ifnotself.check_range_value(cur_value,condition[1][0],condition[1][1]):returnFalseifnotself.check_delay_value(delay_param,condition[1][2]*delay_transform):returnFalse#間隔elifnotself.check_delay_value(delay_param,condition[1]*delay_transform):returnFalse #多個M,N,X在M,N,X時間執行 elifcondition[0]==2:ifnotself.check_multi_value(cur_value,condition[1]):returnFalse #區間A-B elifcondition[0]==3:ifnotself.check_range_value(cur_value,condition[1][0],condition[1][1]):returnFalse#異常else:returnFalse#新增滿足的條件 self.cur_condition+="%s"%cur_valuereturnTruedefon_timer(self,date_time):"""執行檢查"""ifnotself.enable:returnFalse #計算日期差值sub_time=date_time-self.last_check#建立一個新增key，執行檢查可能會成功，但執行過就不執行self.cur_condition=""#分ifnotself._check_condition(self.minutes,date_time.minute,sub_time.seconds,60):self.call_condition=NonereturnFalse#時ifnotself._check_condition(self.hours,date_time.hour,sub_time.seconds,3600):self.call_condition=NonereturnFalse#日ifnotself._check_condition(self.day_of_month,date_time.day,sub_time.days,1):self.call_condition=NonereturnFalse#月ifnotself._check_condition(self.month,date_time.month,self._sub_months(self.last_check,date_time),1):self.call_condition=NonereturnFalse#周ifnotself._check_condition(self.day_of_week,date_time.weekday(),sub_time.days/7,1):self.call_condition=NonereturnFalse#滿足條件則更新檢查時間self.last_check=date_time#只有滿足條件才執行ifself.call_conditionisNoneorself.call_condition!=self.cur_condition:self.call_condition=self.cur_conditionreturnTruereturnFalsedefcall(self):"""執行定時回撥"""try:self.call_able(*self.call_args)exceptExceptionase:print("call",self.call_able,"error:",e)@staticmethoddefcheck_target_value(src_value,des_value):"""檢查指定值"""returnsrc_value==des_value@staticmethoddefcheck_delay_value(src_value,des_value):"""檢查每隔多久"""returnsrc_value>des_value:@staticmethoddefcheck_range_value(src_value,min_value,max_value):"""檢查區間"""returnsrc_value>=min_valueandsrc_value<=max_value@staticmethoddefcheck_multi_value(src_value,values):"""檢查多個值"""returnsrc_valueinvalues@staticmethoddef_sub_months(st,ed):return(ed.year-st.year)*12+(ed.month-st.month)@staticmethoddef_parse_part(value):"""解析單個數據"""value=value.strip()ifvalue=="*":return-1,-1#指定時間一次ifvalue.isdigit():return0,int(value,10)#*/v每隔多久一次fraction_idx=value.find("/")iffraction_idx>=0:denominator=value[fraction_idx+1:]ifnotdenominator.isdigit():return-1,-1ifvalue.startswith("*"):return1,int(denominator,10)else:#特殊區間A-B/N每個多久，且在區間A，B內range_idx=value.find("-")ifrange_idx0:return-1,-1#區間+間隔values=[]range_values=value[:fraction_idx].split("-")iflen(range_values)!=2:return-1,-1#必須是數字，且只存在2個值ifrange_values[0].isdigit()andrange_values[1].isdigit():values.append(int(range_values[0],10))values.append(int(range_values[1],10))values.append(int(denominator,10))return1,values#多個時間點A,B,...,Zdom_idx=value.find(",")ifdom_idx>=0:values=[]forvinvalue.split(","):ifv.isdigit():values.append(int(v,10))return2,values #區間A-Brange_idx=value.find("-")ifrange_idx>=0:values=[]range_values=value.split("-")iflen(range_values)!=2:return-1,-1#必須是數字，且只存在2個值ifrange_values[0].isdigit()andrange_values[1].isdigit():values.append(int(range_values[0],10))values.append(int(range_values[1],10))return3,valuesreturn-1,-1return-1,-1class Timer:    def __init__(self, not_thread=False):"""建立定時器物件"""        self._thread = None        self._timer_data = {}        if not not_thread:            self._thread = threading.Thread(target=self._update, args=())            self._thread.start()    def register(self, time_fmt, call_able, *args):"""註冊定時器"""        timer_data = TimerInfo(time_fmt, call_able, args)        if not timer_data.enable:            return        timer_id = uuid.uuid4()        self._timer_data[timer_id] = timer_data        return timer_id    def _update(self):"""內部更新"""        while self._thread:            self.update(datetime.datetime.now())            time.sleep(5)    def update(self, date_time):"""外部更新使用"""        if not isinstance(date_time, datetime.datetime):raiseTypeError("date_timemustbedatetime")        remove_list = []        for tid, node in self._timer_data.items():            if not node.enable:                remove_list.append(tid)                continue            if node.on_timer(date_time):node.call()        # 刪除已標記刪除的定時器列表        for tid in remove_list:            del self._timer_data[tid]    def remove(self, timer_id):"""移除定時器"""        if timer_id in self._timer_data:            # 只標記            self._timer_data[timer_id].enable = False