在 Python 中，如何让线程休眠到特定时间？答案

【问题标题】：In Python, how can I put a thread to sleep until a specific time?在 Python 中，如何让线程休眠到特定时间？
【发布时间】：2011-01-03 02:51:14
【问题描述】：

我知道我可以通过以下方式使线程休眠一段特定的时间：

time.sleep(NUM)

如何让线程休眠到凌晨 2 点？我必须做数学来确定直到凌晨 2 点的秒数吗？还是有一些库函数？

（是的，我知道 Windows 中的 cron 和等效系统，但我想在 python 中正确休眠我的线程，而不是依赖外部刺激或进程信号。）

【问题讨论】：

我有这个监控 python 脚本，它在许多不同的机器上连续监听套接字。我只想唤醒那个 python 脚本中的一个线程。我不想在启动此脚本的每台机器上都使用 cron。
一般说“在时间X唤醒”的问题是你不能保证内核会在那个时候唤醒线程。计算机可能已关闭、休眠或加载了无法为您节省周期的非常密集的操作。这就是为什么大多数实现不提供这种类型的睡眠。在这种情况下，以秒为单位计算差异可能是最好的方法。
这不是关键任务。如果它只是隔天运行一次，那没关系。它将运行的机器是服务器场中的服务器，因此它们始终处于开启状态并运行多个作业。此外，它不必正好是凌晨 2 点。我只是想让它在大多数人睡着的时候运行。
当然。我可以使用time.gmtime() 来获取格林威治标准时间。我只关心time.localtime()。我不必跨时区协调——无论是用户本地的。
我为我的项目制作的东西...hastebin.com/yufahaciji.py ...抱歉在某些函数中缺少文档字符串耸耸肩

标签： python

【解决方案1】：

这是一个不考虑时钟抖动或时钟调整的半途而废的解决方案。请参阅 cmets 以了解摆脱这种情况的方法。

import time
import datetime

# if for some reason this script is still running
# after a year, we'll stop after 365 days
for i in xrange(0,365):
    # sleep until 2AM
    t = datetime.datetime.today()
    future = datetime.datetime(t.year,t.month,t.day,2,0)
    if t.hour >= 2:
        future += datetime.timedelta(days=1)
    time.sleep((future-t).total_seconds())
    
    # do 2AM stuff

【讨论】：

t.day + (t.hour >= 2) 将是“0000 到 0200 之间”问题的（可能是非 Pythonic）解决方案。另外，我会将sleep 放入循环唤醒中，以防时钟调整或我们早起，但我认为这不是很重要。
顺便说一句，值得注意的是，天真地使用 only seconds 属性可能会导致意外结果。它只包含除以一天的“余数”，可以这么说，所以如果持续时间长于一天，则需要添加.days * 24*3600。在这种情况下显然不是问题，但会抓住不熟悉日期时间对象的奇怪人。
同意。请改用total_seconds()！（它也返回一个浮点数，time.sleep 会接受）
这个解决方案有一个错误 - 未来可能不存在 :) 如果我们在一个月的最后一天怎么办？
@mnowotka，用海绵宝宝的声音，“7 年后” -- timedelta 负责处理日、月和年的滚动。

【解决方案2】：

import pause
from datetime import datetime

pause.until(datetime(2015, 8, 12, 2))

【讨论】：

你能链接到pause 库吗？我在文档中找不到它。这是 Python 3 的新功能吗？
这个包是 14 天前更新的，现在它似乎是可移植的（并且适用于 py3）
@RossRogers 看起来唯一的文档是源中的文档字符串：github.com/jgillick/python-pause/blob/master/pause/…
暂停包文档链接：pypi.org/project/pause
这个库很棒！简单但使用有效的对数递减睡眠循环。

【解决方案3】：

一种可能的方法是睡一个小时。每隔一小时，检查时间是否在半夜。如果是这样，请继续您的操作。如果没有，请再睡一个小时并继续。

如果用户要在中午更改时钟，这种方法将反映该更改。虽然它需要更多的资源，但它应该可以忽略不计。

【讨论】：

我更喜欢这种方法，因为它很简单；它不太可能出现边缘情况。唯一的困难是确保你没有运行它两次，例如如果脚本重新启动。

【解决方案4】：

另一种方法，使用sleep，以对数方式减少超时。

def wait_until(end_datetime):
    while True:
        diff = (end_datetime - datetime.now()).total_seconds()
        if diff < 0: return       # In case end_datetime was in past to begin with
        time.sleep(diff/2)
        if diff <= 0.1: return

基于@MZA 的回答和@Mads Y 的评论

【讨论】：

【解决方案5】：

我尝试了“暂停”包。它不适用于 Python 3.x。从 pause 包中，我提取了等待特定日期时间所需的代码，并进行了以下定义。

def wait_until(execute_it_now):
    while True:
        diff = (execute_it_now - datetime.now()).total_seconds()
        if diff <= 0:
            return
        elif diff <= 0.1:
            time.sleep(0.001)
        elif diff <= 0.5:
            time.sleep(0.01)
        elif diff <= 1.5:
            time.sleep(0.1)
        else:
            time.sleep(1)

【讨论】：

怎么做 sleep((execute_it_now-datetime.today()).total_seconds()/2)？除非我监督了什么，否则这将以对数方式减少睡眠并且非常精确。
pause 的当前版本，v0.3 Oct 2020，非常适合 Python 3.10。

【解决方案6】：

略偏离原问题的重点：

即使您不想乱用 crontabs，如果您可以将 python 脚本安排到这些主机，您可能有兴趣安排 anacron 任务？ anacron 与 cron 的主要区别在于它不依赖计算机连续运行。根据系统配置，即使是此类用户计划的任务，您也可能需要管理员权限。

Ubuntu 人员提供了一个类似的、更现代的工具：http://upstart.ubuntu.com/ 这甚至还没有所需的功能。但安排作业和替换 anacron 是一项计划中的功能。由于它用作 Ubuntu 默认 initd 替代品，因此它具有相当大的吸引力。（我不隶属于该项目）

当然，使用已经提供的答案，您可以将相同的功能编码到您的 python 脚本中，它可能更适合您的情况。

不过，对于其他人来说，anacron 或类似的现有系统可能是更好的解决方案。 anacron 预装在许多当前的 linux 发行版上（Windows 用户存在可移植性问题）。

维基百科提供了一个指针页面：https://en.wikipedia.org/wiki/Anacron

如果您确实选择了 python 版本，我会查看异步方面，并确保即使时间改变（夏令时等）脚本也能正常工作，正如其他人已经评论的那样。与其等到预先计算好的未来，我总是最多等一小时，然后重新检查时间。即使在移动嵌入式系统上，投入的计算周期也应该可以忽略不计。

【讨论】：

【解决方案7】：

适应这个：

from datetime import datetime, timedelta
from time import sleep

now = datetime.utcnow
to = (now() + timedelta(days = 1)).replace(hour=1, minute=0, second=0)
sleep((to-now()).seconds)

【讨论】：

欢迎来到 Stackoverflow，尤利安。已经有很多其他的答案了。你能详细说明你的不同吗？在某些情况下还是在所有情况下更好？

【解决方案8】：

稍微更通用的解决方案（基于Ross Rogers'），以防您也想增加分钟数。

def sleepUntil(self, hour, minute):
    t = datetime.datetime.today()
    future = datetime.datetime(t.year, t.month, t.day, hour, minute)
    if t.timestamp() > future.timestamp():
        future += datetime.timedelta(days=1)
    time.sleep((future-t).total_seconds())

【讨论】：

【解决方案9】：

我知道这已经很晚了，但我想发布一个答案（受标记答案的启发），考虑到系统可能 - 不正确 - 所需的时区 + 包括如何为想知道如何做的人做这个线程。

它看起来很大，因为我正在评论每一步来解释逻辑。

import pytz #timezone lib
import datetime
import time
from threading import Thread

# using this as I am, check list of timezone strings at:
## https://en.wikipedia.org/wiki/List_of_tz_database_time_zones
TIMEZONE = pytz.timezone("America/Sao_Paulo")

# function to return desired seconds, even if it's the next day
## check the bkp_time variable (I use this for a bkp thread)
## to edit what time you want to execute your thread
def get_waiting_time_till_two(TIMEZONE):
    # get current time and date as our timezone
    ## later we remove the timezone info just to be sure no errors
    now = datetime.datetime.now(tz=TIMEZONE).replace(tzinfo=None)
    curr_time = now.time()
    curr_date = now.date()

    # Make 23h30 string into datetime, adding the same date as current time above
    bkp_time = datetime.datetime.strptime("02:00:00","%H:%M:%S").time()
    bkp_datetime = datetime.datetime.combine(curr_date, bkp_time)

    # extract the difference from both dates and a day in seconds
    bkp_minus_curr_seconds = (bkp_datetime - now).total_seconds()
    a_day_in_seconds = 60 * 60 * 24

    # if the difference is a negative value, we will subtract (- with + = -)
    # it from a day in seconds, otherwise it's just the difference
    # this means that if the time is the next day, it will adjust accordingly
    wait_time = a_day_in_seconds + bkp_minus_curr_seconds if bkp_minus_curr_seconds < 0 else bkp_minus_curr_seconds

    return wait_time

# Here will be the function we will call at threading
def function_we_will_thread():
    # this will make it infinite during the threading
    while True:
        seconds = get_waiting_time_till_two(TIMEZONE)
        time.sleep(seconds)
        # Do your routine

# Now this is the part where it will be threading
thread_auto_update = Thread(target=function_we_will_thread)
thread_auto_update.start()

【讨论】：

【解决方案10】：

它只需要一个非常基本的库。

import time

sleep_until = 'Mon Dec 25 06:00:00 2020' # String format might be locale dependent.
print("Sleeping until {}...".format(sleep_until))
time.sleep(time.mktime(time.strptime(sleep_until)) - time.time())

time.strptime() 从字符串 -> struct_time 元组解析时间。如果您将 strptime() 解析格式字符串作为第二个参数，则字符串可以采用不同的格式。例如。 time.strptime("12/25/2020 02:00AM", "%m/%d/%Y %I:%M%p")
time.mktime() 将 struct_time -> 纪元时间转换为秒。
time.time() 以秒为单位给出当前纪元时间。
从前者中减去后者，即可得到所需的睡眠时间（以秒为单位）。
sleep() 数量。

如果您只想睡到下一个凌晨 2 点（可能是今天或明天），您需要一个 if 语句来检查今天的时间是否已经过去。如果有，请将唤醒设置为第二天。

import time

sleep_until = "02:00AM" # Sets the time to sleep until.

sleep_until = time.strftime("%m/%d/%Y " + sleep_until, time.localtime()) # Adds todays date to the string sleep_until.
now_epoch = time.time() #Current time in seconds from the epoch time.
alarm_epoch = time.mktime(time.strptime(sleep_until, "%m/%d/%Y %I:%M%p")) # Sleep_until time in seconds from the epoch time.

if now_epoch > alarm_epoch: #If we are already past the alarm time today.
    alarm_epoch = alarm_epoch + 86400 # Adds a day worth of seconds to the alarm_epoch, hence setting it to next day instead.

time.sleep(alarm_epoch - now_epoch) # Sleeps until the next time the time is the set time, whether it's today or tomorrow.

【讨论】：

【解决方案11】：

from datetime import datetime
import time, operator
time.sleep([i[0]*3600 + i[1]*60 for i in [[H, M]]][0] - [i[0]*3600 + i[1]*60 for i in [map(int, datetime.now().strftime("%H:%M").split(':'))]][0])

【讨论】：

【解决方案12】：

除了使用 wait() 函数，您可以使用 while 循环检查是否已达到指定日期：


if datetime.datetime.utcnow() > next_friday_10am:
    # run thread or whatever action
    next_friday_10am = next_friday_10am()
    time.sleep(30)

def next_friday_10am():
    for i in range(7):
        for j in range(24):
            for k in range(60):
                if (datetime.datetime.utcnow() + datetime.timedelta(days=i)).weekday() == 4:
                    if (datetime.datetime.utcnow() + datetime.timedelta(days=i, hours=j)).hour == 8:
                        if (datetime.datetime.utcnow() + datetime.timedelta(days=i, hours=j, minutes=k)).minute == 0:
                            return datetime.datetime.utcnow() + datetime.timedelta(days=i, hours=j, minutes=k)

仍然有时间检查线程每 30 秒检查一次条件，因此需要的计算量比等待时多，但这是使其工作的一种方法。

【讨论】：