Python 在特定时间启动进程并在特定时间结束

Question

这是一个示例 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函数定义为一个全局对象。但它一直在运行。我想在特定时间开始该过程并在特定时间结束它。我的代码有什么问题？

Answer 1

这段代码有几个问题。首先，当在线程内部循环并且没有条件命中时，添加睡眠以节省 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

请记住，您需要定义线程开启和关闭的确切时间间隔。有条件地涵盖所有这些情况并采取相应的行动。随着您的项目变得更加复杂，稍后可能需要互斥锁。

Answer 2

我已经为您实现了这个解决方案：

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__函数初始化程序。