欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

python異步實現(xiàn)定時任務(wù)和周期任務(wù)的方法

 更新時間:2019年06月29日 10:14:45   作者:狡猾的皮球  
今天小編就為大家分享一篇python異步實現(xiàn)定時任務(wù)和周期任務(wù)的方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧

一. 如何調(diào)用

def f1(arg1, arg2):
  print('f1', arg1, arg2)
 
 
def f2(arg1):
  print('f2', arg1)
 
 
def f3():
  print('f3')
 
 
def f4():
  print('周期任務(wù)', int(time.time()))
 
 
timer = TaskTimer()
# 把任務(wù)加入任務(wù)隊列
timer.join_task(f1, [1, 2], timing=15.5) # 每天15:30執(zhí)行
timer.join_task(f2, [3], timing=14) # 每天14:00執(zhí)行
timer.join_task(f3, [], timing=15) # 每天15:00執(zhí)行
timer.join_task(f4, [], interval=10) # 每10秒執(zhí)行1次
# 開始執(zhí)行(此時才會創(chuàng)建線程)
timer.start()

f1~f4是我們需要定時執(zhí)行的函數(shù)。

首先創(chuàng)建TaskTimer對象(TaskTimer的代碼在下面)。調(diào)用join_task函數(shù),把需要執(zhí)行的函數(shù)加入到任務(wù)隊列。最后調(diào)用start,任務(wù)就開始執(zhí)行了。

join_task參數(shù):

fun:需要執(zhí)行的函數(shù)

arg:fun的參數(shù),如果沒有就傳一個空列表

interval:如果有此參數(shù),說明任務(wù)是周期任務(wù),單位為秒(注意interval最少5秒)

timing:如果有此參數(shù),說明任務(wù)是定時任務(wù),單位為時

注意:interval和timing只能選填1個

二. 源碼

import datetime
import time
from threading import Thread
from time import sleep
 
 
class TaskTimer:
  __instance = None
 
  def __new__(cls, *args, **kwargs):
    """
    單例模式
    """
    if not cls.__instance:
      cls.__instance = object.__new__(cls)
    return cls.__instance
 
  def __init__(self):
    if not hasattr(self, 'task_queue'):
      setattr(self, 'task_queue', [])
 
    if not hasattr(self, 'is_running'):
      setattr(self, 'is_running', False)
 
  def write_log(self, level, msg):
    cur_time = datetime.datetime.now()
    with open('./task.log', mode='a+', encoding='utf8') as file:
      s = "[" + str(cur_time) + "][" + level + "]  " + msg
      print(s)
      file.write(s + "\n")
 
  def work(self):
 
    """
    處理任務(wù)隊列
    """
    while True:
      for task in self.task_queue:
        if task['interval']:
          self.cycle_task(task)
        elif task['timing']:
          self.timing_task(task)
 
      sleep(5)
 
  def cycle_task(self, task):
    """
    周期任務(wù)
    """
    if task['next_sec'] <= int(time.time()):
      try:
        task['fun'](*task['arg'])
        self.write_log("正常", "周期任務(wù):" + task['fun'].__name__ + " 已執(zhí)行")
      except Exception as e:
        self.write_log("異常", "周期任務(wù):" + task['fun'].__name__ + " 函數(shù)內(nèi)部異常:" + str(e))
      finally:
        task['next_sec'] = int(time.time()) + task['interval']
 
  def timing_task(self, task):
    """
    定時任務(wù)
    """
    # 今天已過秒數(shù)
    today_sec = self.get_today_until_now()
 
    # 到了第二天,就重置任務(wù)狀態(tài)
    if task['today'] != self.get_today():
      task['today'] = self.get_today()
      task['today_done'] = False
 
    # 第一次執(zhí)行
    if task['first_work']:
      if today_sec >= task['task_sec']:
        task['today_done'] = True
        task['first_work'] = False
      else:
        task['first_work'] = False
 
    # 今天還沒有執(zhí)行
    if not task['today_done']:
      if today_sec >= task['task_sec']: # 到點了,開始執(zhí)行任務(wù)
        try:
          task['fun'](*task['arg'])
          self.write_log("正常", "定時任務(wù):" + task['fun'].__name__ + " 已執(zhí)行")
        except Exception as e:
          self.write_log("異常", "定時任務(wù):" + task['fun'].__name__ + " 函數(shù)內(nèi)部異常:" + str(e))
        finally:
          task['today_done'] = True
          if task['first_work']:
            task['first_work'] = False
 
  def get_today_until_now(self):
    """
    獲取今天凌晨到現(xiàn)在的秒數(shù)
    """
    i = datetime.datetime.now()
    return i.hour * 3600 + i.minute * 60 + i.second
 
  def get_today(self):
    """
    獲取今天的日期
    """
    i = datetime.datetime.now()
    return i.day
 
  def join_task(self, fun, arg, interval=None, timing=None):
    """
    interval和timing只能存在1個
    :param fun: 你要調(diào)用的任務(wù)
    :param arg: fun的參數(shù)
    :param interval: 周期任務(wù),單位秒
    :param timing: 定時任務(wù),取值:[0,24)
    """
    # 參數(shù)校驗
    if (interval != None and timing != None) or (interval == None and timing == None):
      raise Exception('interval和timing只能選填1個')
 
    if timing and not 0 <= timing < 24:
      raise Exception('timing的取值范圍為[0,24)')
 
    if interval and interval < 5:
      raise Exception('interval最少為5')
 
    # 封裝一個task
    task = {
      'fun': fun,
      'arg': arg,
      'interval': interval,
      'timing': timing,
    }
    # 封裝周期或定時任務(wù)相應(yīng)的參數(shù)
    if timing:
      task['task_sec'] = timing * 3600
      task['today_done'] = False
      task['first_work'] = True
      task['today'] = self.get_today()
    elif interval:
      task['next_sec'] = int(time.time()) + interval
 
    # 把task加入任務(wù)隊列
    self.task_queue.append(task)
 
    self.write_log("正常", "新增任務(wù):" + fun.__name__)
 
  def start(self):
    """
    開始執(zhí)行任務(wù)
    返回線程標(biāo)識符
    """
    if not self.is_running:
      thread = Thread(target=self.work)
 
      thread.start()
 
      self.is_running = True
 
      self.write_log("正常", "TaskTimer已開始運(yùn)行!")
 
      return thread.ident
 
    self.write_log("警告", "TaskTimer已運(yùn)行,請勿重復(fù)啟動!")

以上這篇python異步實現(xiàn)定時任務(wù)和周期任務(wù)的方法就是小編分享給大家的全部內(nèi)容了,希望能給大家一個參考,也希望大家多多支持腳本之家。

相關(guān)文章

  • Python中函數(shù)參數(shù)調(diào)用方式分析

    Python中函數(shù)參數(shù)調(diào)用方式分析

    這篇文章主要介紹了Python中函數(shù)參數(shù)調(diào)用方式,結(jié)合實例形式分析了Python函數(shù)參數(shù)定義與使用的四種常見操作方法,需要的朋友可以參考下
    2018-08-08
  • 淺析如何在Python中使用結(jié)構(gòu)模式匹配

    淺析如何在Python中使用結(jié)構(gòu)模式匹配

    在Python 3.10中引入了模式匹配語法,允許我們在應(yīng)用程序中使用強(qiáng)大的新編程技術(shù)進(jìn)行決策,下面我們就來講講如何在Python中使用結(jié)構(gòu)模式匹配吧
    2023-08-08
  • python實現(xiàn)月食效果實例代碼

    python實現(xiàn)月食效果實例代碼

    在本文里小編給大家整理了關(guān)于python實現(xiàn)月食效果的相關(guān)實例內(nèi)容以及對應(yīng)代碼,有興趣的朋友們學(xué)習(xí)下。
    2019-06-06
  • python中@property和property函數(shù)常見使用方法示例

    python中@property和property函數(shù)常見使用方法示例

    這篇文章主要介紹了python中@property和property函數(shù)常見使用方法,結(jié)合實例形式分析了Python @property和property函數(shù)功能、使用方法及相關(guān)操作注意事項,需要的朋友可以參考下
    2019-10-10
  • python循環(huán)語句的使用方法

    python循環(huán)語句的使用方法

    這篇文章主要介紹了python循環(huán)語句的使用方法,文章主要介紹python循環(huán)相關(guān)內(nèi)容有?for循環(huán)語句、循環(huán)嵌套、break結(jié)束循環(huán)、continue跳過本次循環(huán)實現(xiàn)內(nèi)容,需要的小伙伴可以參考一下
    2022-04-04
  • python實現(xiàn)電子書翻頁小程序

    python實現(xiàn)電子書翻頁小程序

    這篇文章主要為大家詳細(xì)介紹了python實現(xiàn)電子書翻頁小程序,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2019-07-07
  • python基礎(chǔ)入門學(xué)習(xí)筆記(Python環(huán)境搭建)

    python基礎(chǔ)入門學(xué)習(xí)筆記(Python環(huán)境搭建)

    這篇文章主要介紹了python基礎(chǔ)入門學(xué)習(xí)筆記,這是開啟學(xué)習(xí)python基礎(chǔ)知識的第一篇,夯實Python基礎(chǔ),才能走的更遠(yuǎn),感興趣的小伙伴們可以參考一下
    2016-01-01
  • 以SortedList為例詳解Python的defaultdict對象使用自定義類型的方法

    以SortedList為例詳解Python的defaultdict對象使用自定義類型的方法

    這篇文章主要介紹了以SortedList為例詳解Python的defaultdict對象使用自定義類型的方法,文章圍繞主題展開詳細(xì)的內(nèi)容介紹,具有一定的參考價值,需要的朋友可以參考一下
    2022-07-07
  • LyScript實現(xiàn)繞過反調(diào)試保護(hù)的示例詳解

    LyScript實現(xiàn)繞過反調(diào)試保護(hù)的示例詳解

    LyScript插件中內(nèi)置的方法可實現(xiàn)各類反調(diào)試以及屏蔽特定API函數(shù)的功能,這類功能在應(yīng)對病毒等惡意程序時非常有效。本文為大家提供了LyScript實現(xiàn)繞過反調(diào)試保護(hù)的示例代碼,感興趣的可以了解一下
    2022-08-08
  • keras K.function獲取某層的輸出操作

    keras K.function獲取某層的輸出操作

    這篇文章主要介紹了keras K.function獲取某層的輸出操作,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2020-06-06

最新評論