這是一個示例 python 代碼，每 5 秒運行一個進程，直到凌晨 1 點。def run_check(): mythrd =threading.Timer(5.0, run_check) mythrd.start() time = datetime.now() domystuff() if time.hour == 1: mythrd.cancel()run_check()我想在特定時間啟動線程。所以我改變了這個代碼如下，mythrd = Nonedef run_check(): while (True): now = datetime.now if now().hour == 16 and now().minute == 30 and now().second == 30: mythrd.start() time.sleep(1) if mythrd.is_alive(): domystuff() if datetime.now().hour == 16 and datetime.now().minute == 31: mythrd.cancel()mythrd =threading.Timer(5.0, run_check)run_check()早些時候，我將mythrd對象保留在run_check函數中。如果我在run_check函數內部初始化，它將將該對象作為一個新對象。它會再次觸發。所以我將mythrd函數定義為一個全局對象。但它一直在運行。我想在特定時間開始該過程并在特定時間結束它。我的代碼有什么問題？
2 回答
MMMHUHU
TA貢獻1834條經驗 獲得超8個贊
這段代碼有幾個問題。首先，當在線程內部循環并且沒有條件命中時，添加睡眠以節省 CPU 使用。其次，由于線程是非確定性的，您不能在確切的時間戳上觸發條件。例如，線程可能會在“熱”時間間隔內跳過它的執行，并且條件永遠不會觸發。
因此，您可以改為檢查當前時間戳是否已超過某個時間戳（以及超過了多少）和/或以其他方式比較時間戳。
但無論如何，在您的代碼中 mythrd 啟動一個線程函數，它將遞歸調用自己......這就是問題所在。
mythrd =threading.Timer(5.0, run_check)
進而
mythrd.start() # will start threading.Timer(5.0, run_check)
因此，當條件now().hour == 16 and now().minute == 30 and now().second == 30觸發時，線程將運行自身的另一個實例。這個實例可能會運行另一個實例等等......
您的意思是實現一個觸發線程，執行run_check任務，然后啟動另一個線程執行實際工作？
這是一個解決方案：
from threading import Thread, Timer
import time
import datetime
keep_checking = True
keep_working = True
def worker_thread():
while keep_working:
pass #do my stuff here and remember to use an eventual sleep
def time_checker():
while keep_checking:
now = datetime.now()
if now.hour == 16 and now.minute == 30 and (
now.second >= 30 and now.second < 31) and (
not mythrd.isAlive()):
mythrd.start()
elif now.minute >= 31 and mythrd.isAlive():
keep_working = False # and let it die instead to kill it
else:
time.sleep(1) # no operation
mythrd = threading.Thread(target=worker_thread)
time_checker() # might turn this into a thread as well
請記住，您需要定義線程開啟和關閉的確切時間間隔。有條件地涵蓋所有這些情況并采取相應的行動。隨著您的項目變得更加復雜，稍后可能需要互斥鎖。
PIPIONE
TA貢獻1829條經驗 獲得超9個贊
我已經為您實現了這個解決方案：
from datetime import datetime as dt
import threading
import time
class test(object):
def __init__(self):
self.run_permission = False
self.start_time = [16,30,30] # Hours, Minutes, Seconds
self.end_time = [18,41,00] # Hours, Minutes, Seconds
self.timer_sched_time = 5 # seconds
threading.Thread(name="time_checker", target=self.check_time, args=(self.start_time, self.end_time,)).start()
threading.Thread(name="scheduled_job", target=self.Timer, args=(self.timer_sched_time, )).start()
while True:
time.sleep(10)
def timer_job(self, unix_time, human_time):
print "Unix time: %s Human time: %s \n" % (unix_time, human_time)
def Timer(self, delay):
while True:
try:
time_remaining = delay-time.time()%delay
time.sleep(time_remaining)
unix_time = str(int(round(time.time()*1000)))
human_time = str(dt.now())
if(self.run_permission):
self.timer_job(unix_time, human_time)
except Exception, ex:
raise ex
def check_time(self, start_execution_time, end_execution_time):
while True:
now = dt.now()
if(now.hour >= start_execution_time[0] and now.minute >= start_execution_time[1] and now.second >= start_execution_time[2]):
self.run_permission = True
if(now.hour >= end_execution_time[0] and now.minute >= end_execution_time[1] and now.second >= end_execution_time[2]):
self.run_permission = False
test()
首先導入您需要的庫：
from datetime import datetime as dt
import threading
import time
創建一個名為 test 的類：
class test(object):
定義__init__，當您使用test()以下命令調用它時，它將在創建類 test 時執行：
def __init__(self):
self.run_permission = False
self.start_time = [16,30,30] # Hours, Minutes, Seconds
self.end_time = [18,41,00] # Hours, Minutes, Seconds
self.timer_sched_time = 5 # seconds
threading.Thread(name="time_checker", target=self.check_time, args=(self.start_time, self.end_time,)).start()
threading.Thread(name="scheduled_job", target=self.Timer, args=(self.timer_sched_time, )).start()
while True:
time.sleep(10)
該變量run_permission是一個布爾值，用作檢查是否可以在計時器中執行作業的標志。
該start_time變量中設置的開始時間check_time是寫程序run_permission變量，True如果時間主要或等于start_time
該end_time變量設定的停止時間check_time是寫程序run_permission變量設置為false如果時間主要或等于end_time
該timer_sched_time變量設置定時器的延遲，計算睡眠時間需要延遲，它每 5 秒授予完美的時間執行：17.00.00 - 17.00.05 - 17.00.10 - 17.00.15 等等
這兩行啟動check_time作業和Timer作業的線程：
threading.Thread(name="time_checker", target=self.check_time, args=(self.start_time, self.end_time,)).start()
threading.Thread(name="scheduled_job", target=self.Timer, args=(self.timer_sched_time, )).start()
這兩行保持主線程運行，但睡眠主線程以減少資源消耗：
while True:
time.sleep(10)
該函數timer_job僅以 Unix 格式和人類可讀格式打印時間：
def timer_job(self, unix_time, human_time):
print "Unix time: %s Human time: %s \n" % (unix_time, human_time)
要翻譯 unix 格式，您可以使用epoch 轉換器網站。
該Timer函數計算睡眠時間以將執行授予完美的時間，以 unix 和人類格式獲取時間戳并將其傳遞給timer_job如果run_permission變量設置為True：
def Timer(self, delay):
while True:
try:
time_remaining = delay-time.time()%delay
time.sleep(time_remaining)
unix_time = str(int(round(time.time()*1000)))
human_time = str(dt.now())
if(self.run_permission):
self.timer_job(unix_time, human_time)
except Exception, ex:
raise ex
check_time 函數使用參數start_execution_time 并將end_execution_time變量設置run_permission為TrueorFalse是否遵守時間執行條件（因此如果遵守小時/分鐘/秒范圍）：
def check_time(self, start_execution_time, end_execution_time):
while True:
now = dt.now()
if(now.hour >= start_execution_time[0] and now.minute >= start_execution_time[1] and now.second >= start_execution_time[2]):
self.run_permission = True
if(now.hour >= end_execution_time[0] and now.minute >= end_execution_time[1] and now.second >= end_execution_time[2]):
self.run_permission = False
如前所述test()，在文件末尾調用測試類并使用該__init__函數初始化程序。
添加回答
舉報
0/150
提交
取消