Python APScheduler - AsyncIOScheduler 如何工作？

Question

我很难理解AsyncIOScheduler是如何工作的，它是如何非阻塞的？

如果我的工作正在执行阻塞功能，会AsyncIOScheduler阻塞吗？

如果我使用AsyncIOSchedulerwithThreadPoolExecutor呢？这是如何运作的？我可以等待作业执行吗？

Answer 1

使用一些互联网资源，我发现了一些有用的事实。希望它会帮助你。

一个典型的 APScheduler 实例包含数十个执行常规 Python 函数的作业。APScheduler 实例可以调度的作业数量没有限制；它只取决于机器的实际负载。默认情况下，APScheduler 将所有作业存储在内存中。如果您希望您的作业在进程重新启动后继续存在并从上次触发时继续触发，您可以将这些作业存储在数据库中，例如任何 RDBMS、Redis、MongoDB 等。

根据您的应用程序的运行方式，它可以作为线程、异步任务或其他方式运行。初始化时，除非您将 Python 函数添加为作业，否则 APScheduler 不会执行任何操作。添加所有作业后，您需要“启动”调度程序。有关如何使用 APScheduler 的简单示例，这里是一段有效的代码。

from urllib.request import urlopen
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job("interval", seconds=10)
def keep_warm():
    urlopen("https://enqueuezero.com", timeout=10)
    
scheduler.start()

这确保每 10 秒请求一个 URL。该程序作为阻塞进程运行。与你的应用程序如果你想并存的他们，你可以考虑使用BackgroundScheduler，AsyncIOScheduler等等。

这里有一些代码片段BackgroundScheduler。

from datetime import datetime
import time
import os

from apscheduler.schedulers.background import BackgroundScheduler


def tick():
    print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    try:
        # This is here to simulate application activity (which keeps the main thread alive).
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        # Not strictly necessary if daemonic mode is enabled but should be done if possible
        scheduler.shutdown()

下面的代码将演示如何使用asyncio兼容的调度程序来调度以 3 秒为间隔执行的作业。

import asyncio
import os
from datetime import datetime

from apscheduler.schedulers.asyncio import AsyncIOScheduler


def tick():
    print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
    scheduler = AsyncIOScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    # Execution will block here until Ctrl+C (Ctrl+Break on Windows) is pressed.
    try:
        asyncio.get_event_loop().run_forever()
    except (KeyboardInterrupt, SystemExit):
        pass

您可以使用此链接获得更多有关 apscheduler 的示例！

Answer 2

因此，在 APScheduler 中有 3 个重要组件：

1. 调度器
2. 执行人
3. 数据存储

对于这个问题，只有 1 和 2 是相关的。

调度程序只是根据它们的间隔设置决定何时调用作业，在这种情况下，AsyncIOScheduler它使用 asyncio 使等待期非阻塞。它运行在与主进程相同的进程和线程中。如果您的应用程序已经在 asyncio 循环上运行，这将非常有用，因为它节省了运行新进程/线程的开销。

现在，当需要执行作业时，调用的是 Executor，在这种情况下AsyncIOScheduler，默认情况下是 uses AsyncIOExecutor，它运行在与调度程序相同的线程和进程中，如果作业函数被签名为异步，否则它使用asyncio 的 run_in_executor 在线程池中运行它。

这就引出了最后一个问题，如果我们使用AsyncIOSchedulerwith会发生什么ThreadPoolExecutor？好吧，技术上与使用带有非异步函数的默认 Executor 相同，它将在线程池中运行它，但调度程序将保留在主线程中。

Answer 3

根据文档，在事件循环中执行 AsyncIOScheduler。

它是非阻塞的，因为它只会将自己添加到事件循环中并等待您启动它。

一旦事件循环启动，它将异步运行。

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio

async def job():
    print('hi')

scheduler = AsyncIOScheduler()
scheduler.add_job(job, "interval", seconds=3)

scheduler.start()

asyncio.get_event_loop().run_forever()

输出

Run time of job "job (trigger: interval[0:00:03], next run at: 2020-07-27 14:06:39 -03)" was missed by 0:00:02.542515
hi
hi