Python 在特定时间开始流程并在特定时间结束流程
Python starting a process in a particular time and ending it on a particular time
这是一个示例 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 = None
def 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 函数定义为一个全局对象。但它保持 运行ning。我想在特定时间开始该过程并在特定时间结束它。我的代码有什么问题？
这段代码有几个问题。
首先，当在线程内循环且没有条件命中时，添加一个睡眠以节省 CPU 使用。
其次，由于线程是不确定的，您不能在确切的时间戳上触发条件。例如，线程可能会在 "hot" 时间间隔内跳过它的执行，并且条件永远不会触发。
因此，您可以改为检查当前时间戳是否超过某个时间戳（以及超过多少）and/or，否则比较时间戳。
但无论如何，在您的代码中，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
请记住，您需要定义线程开启和关闭的确切时间间隔。涵盖所有有条件的情况并采取相应行动。随着您的项目变得更加复杂，稍后可能需要互斥锁。
我已经为你实现了这个解决方案：
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
创建一个 class 命名测试：
class test(object):
定义__init__，当你使用test():[调用它时，它将在创建class测试时执行=47=]
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 例程的开始时间，如果时间主要或等于 start_time
end_time 变量设置 check_time 例程的停止时间，如果时间主要或等于 end_time[，则将 run_permission 变量写入 False =47=]
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 converter website。
如果 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 设置为 True 或 False 如果遵守时间执行条件（因此，如果遵守 hours/minutes/seconds 范围）：
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() 调用测试 class 并使用 __init__ 函数初始化程序。
这是一个示例 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 = None
def 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 函数定义为一个全局对象。但它保持 运行ning。我想在特定时间开始该过程并在特定时间结束它。我的代码有什么问题？
这段代码有几个问题。 首先，当在线程内循环且没有条件命中时，添加一个睡眠以节省 CPU 使用。 其次，由于线程是不确定的，您不能在确切的时间戳上触发条件。例如，线程可能会在 "hot" 时间间隔内跳过它的执行，并且条件永远不会触发。
因此，您可以改为检查当前时间戳是否超过某个时间戳（以及超过多少）and/or，否则比较时间戳。
但无论如何，在您的代码中，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
请记住，您需要定义线程开启和关闭的确切时间间隔。涵盖所有有条件的情况并采取相应行动。随着您的项目变得更加复杂，稍后可能需要互斥锁。
我已经为你实现了这个解决方案：
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
创建一个 class 命名测试：
class test(object):
定义__init__，当你使用test():[调用它时，它将在创建class测试时执行=47=]
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 例程的开始时间，如果时间主要或等于 start_time
end_time 变量设置 check_time 例程的停止时间，如果时间主要或等于 end_time[，则将 run_permission 变量写入 False =47=]
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 converter website。
如果 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 设置为 True 或 False 如果遵守时间执行条件（因此，如果遵守 hours/minutes/seconds 范围）：
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() 调用测试 class 并使用 __init__ 函数初始化程序。