python - 使用内置 sched 模块的非阻塞 scheduler.run() 方法？

Question

我想了解如何使用可选参数blocking在方法 scheduler.run(blocking=True) .任何实际/现实世界的例子都会非常有帮助。

根据我迄今为止所做的研究，blocking 的意图可选参数用于非阻塞或异步应用程序[1][2]。下面是 schduler 的主运行循环(来自 python 3.6 库 sched.py )。通过代码，我注意到每当 blocking设置为 False , 立即返回目标时间和当前时间之间的时间差，除非目标时间已经过去，在这种情况下将执行操作。

while True:
    with lock:
        if not q:
            break
        time, priority, action, argument, kwargs = q[0]
        now = timefunc()
        if time > now:
            delay = True
        else:
            delay = False
            pop(q)
    if delay:
        if not blocking:
            return time - now
        delayfunc(time - now)
    else:
        action(*argument, **kwargs)
        delayfunc(0)   # Let other threads run

在我看来，非阻塞设计要求我继续运行调度程序，直到队列干净为止。因此，我正在考虑自己维护一个任务队列并继续推送 scheduler.run。任务进入队列(如下面的代码)。这是一个理想的设计吗？使用非阻塞调度程序的正确方法是什么？

def action():
    print('action at: ', datetime.now())

if __name__ == '__main__':
    s = sched.scheduler(time.time)
    target_time = datetime.now() + timedelta(seconds=5)
    s.enterabs(target_time.timestamp(), 1, action)
    run = functools.partial(s.run, blocking=False)
    taskq = deque()
    taskq.append(run)
    while taskq:
        task = taskq.popleft()
        result = task()
        print(result)
        if result:
            taskq.append(run)
            time.sleep(1)

    print('end tasks')

[1] What’s New In Python 3.3

[2] Issue13449: sched - provide an "async" argument for run() method

Answer 1

老问题，但我刚刚实现了一些非常有效地使用非阻塞版本的东西。

当blocking = True在 sched.scheduler.run ，它将调用 delayfunc 的时间差，直到下一个事件。

如果您的应用程序位于 t = 0，这可能是不可取的。 ，安排事件A对于 t = 10 ，但另一个线程，在 t = 1 ，安排事件B对于 t = 5 .在这种情况下，

s = sched.scheduler(time.time)
# Spawn threads which enter A and B into s
while True:
    s.run(True)

如果你的主线程只是调用 sched.scheduler.run(blocking=True)在一个循环中，在 t = 0它将调用 delayfunc(10)因为它只看到距离 A 还剩 10 个时间单位.主线程直到 t = 10 才会唤醒，此时它会看到它错过了 B , 运行 B延迟 5 个时间单位，然后运行 ​​ A在 B 之后.

为了解决这个问题，您可以将主线程更改为如下所示:

s = sched.scheduler(time.time)
# Spawn threads which enter A and B into s
while True:
    next_ev = s.run(False)
    if next_ev is not None:
        time.sleep(min(1, next_ev))
    else:
        time.sleep(1)

此代码将 catch 所有当前事件，然后休眠直到下一个事件，或者如果没有下一个事件或下一个事件太远，将休眠 1 秒。

理想情况下，调度程序将使用条件变量来实现，以判断新事件是否到达优先级队列的前面，并且它可以等待该变量而不是仅仅休眠直到下一个事件。这将是最有效和最准确的时间。