前言

前面我们分享了一个关于pyee库的使用，不过它注重的事件的分发处理。而且对应异步的应用上来说，它没有使用到多少。接下里我们再学习一个异步库aiojobs。它是一个用于管理异步后台任务的作业调度程序。这个对于我们的一些web应用中做相关后台任务，其实还是非常方便的，至少我们可以给我们多一个选择，也就是我们可以不依赖于FastAPI框架自带的后台任务。我们可以自己定义一个全局异步后台任务管理器，进行相关后台任务的处理。

aiojobs的应用

我们首先有一个全貌认识的话，当然是从提供的官方文档开始，所以我们直接解读我们的官方文档。

首先我们先安装它：

pip install aiojobs

然后我直接的抄写官方的文档的示例：

import asyncio
import aiojobs

async def coro(timeout):
    """
    异步函数：等待一段时间

    参数：
    - timeout (float): 等待的时长

    返回值：
    无
    """
    await asyncio.sleep(timeout)

async def main():
    """
    主函数：调度器测试

    参数：
    无

    返回值：
    无
    """
    scheduler = aiojobs.Scheduler()
    for i in range(100):
        # 生成任务
        await scheduler.spawn(coro(i/10))

    await asyncio.sleep(5.0)
    # 并不是所有已调度的任务都已完成

    # 温和地关闭已生成的任务
    await scheduler.close()

asyncio.run(main())

前面我们说过了aiojobs是一个基于asyncio实现的任务调度库，方便管理和控制协程任务。

在上面的代码中，首先我们定义了一个名为coro的异步函数，它接受一个参数timeout表示等待的时间（单位秒）。然后在我们的函数内部，通过调用asyncio.sleep(timeout)使得当前协程挂起并等待指定时长后恢复执行。

接着我们在主函数main()入口中：

1. 创建了一个aiojobs.Scheduler实例scheduler，它是任务调度器。

2. 然后使用for循环生成100个任务，并使用使用await scheduler.spawn对每个coro函数的进行一次调用将任务添加到调度器中，使其异步执行。（ 使用scheduler.spawn()方法启动一个作业，添加后将开始异步函数的执行）

3. 接着在主进程中，我们等待5秒（await asyncio.sleep(5.0)），此时并不是所有已调度的任务都已完成。

4. 然后调用await scheduler.close()来关闭调度器并等待所有已生成的任务完成。这是一种“温和”的关闭方式，即确保所有任务有机会正常结束而不是立即终止。

从上面我们可以了解到aiojobs中的，两个核心概念：Job和Scheduler。

• Job是围绕asyncio任务的包装器，可以启动异步函数的执行，等待结果/异常，并关闭。

• Scheduler是用于存放已启动的Job的容器，隐含地限制了并发数，也可以进行异常的处理。

• aiojobs的Scheduler默认有一个并发作业的数量限制（默认为100）。如果达到这个限制，新的作业将被加入到待处理列表中。一旦一个正在运行的作业停止，下一个待处理的作业将被执行

• 上面的截图，我们可以看得出来，Scheduler 类被定义为 Collection[Job[object]]。也就是说我们的Scheduler 类是一个集合，其元素类型为 Job[object]类型的对象。在这里Job 是一个泛型类型，它接受一个类型参数 object，这意味着 Job 中的对象可以是任何类型。

• 上面的截图中，我们的Job 是一个泛型类，它表示一个异步任务。在这个上下文中，Job 类是 Scheduler 类中管理的一个对象，用于封装异步函数和其执行。

• aiojobs还提供了优雅的关闭功能。所有已启动的作业都会被停止和关闭。关闭操作有一个超时设置（默认为0.1秒），如果在超时时间内作业无法关闭，将会记录一条日志

• 调度中的spawn()方法用于创建并启动新的作业。如果并发限制被超过，新作业可能会被推入待处理列表中。

• 调度中的close()方法用于关闭Scheduler及其所有作业，如果在超时时间内某个作业无法关闭，将会记录一条日志 。

• exception_handler是一个回调函数，用于处理作业中未明确等待的异常

关于Scheduler中的spawn()方法

我们拿里面的源码分析一下：

    async def spawn(self, coro: Coroutine[object, object, _T]) -> Job[_T]:
        # 判断是否已关闭
        if self._closed:
            raise RuntimeError("在关闭后又调度了一个新的任务")
        # 创建一个新的任务
        job = Job(coro, self)
        # 判断是否满足开始任务的条件
        should_start = self._limit is None or self.active_count < self._limit
        if should_start:
            # 启动任务
            job._start()
        else:
            try:
                # 等待队列中有空余的位置
                await self._pending.put(job)
            except asyncio.CancelledError:
                # 如果任务被取消，则关闭任务
                await job.close()
                raise
        # 将任务加入任务集合中
        self._jobs.add(job)
        # 返回任务对象
        return job

从上面的代码我们也看得出来：当我们添加新的任务协程对象到该spawn方法的时候：

1. 检查调度器状态：首先检查当前调度器是否已关闭，如果已关闭，则抛出RuntimeError异常，提示不能在关闭后调度新的任务。

2. 接着会把使用传入的协程对象coro转为为一个job的任务实例，并将其与当前调度器关联，创建出新的任务。

• 如果无限制或当前活跃任务数量未达到限制值，则可以立即启动新任务。启动任务：若满足启动条件，调用job._start()方法启动新任务。
• 如果不满足立即启动条件，则将新任务加入到待处理任务队列（self._pending）中，并等待有空闲位置。在此过程中，如果遭遇asyncio.CancelledError异常，表示当前操作被取消，此时会关闭新任务，并重新抛出取消异常。

3. 判断启动条件：根据调度器的限制（self._limit）和当前活跃任务数量（self.active_count）判断是否满足立即启动新任务的条件。

4. 最后添加任务至集合：无论任务是否立即启动，最终都会将其添加到调度器管理的所有任务集合（self._jobs）中。

5. 最终返回任务实例：最后返回创建好的任务实例job，便于我们在外部对任务进行后续的操作和管理。

关于Job中的spawn()方法

当我们的任务可以被调度启动的时候，它执行的是Job中的_start方法，我们看一下_start的访问的内部逻辑，如下所示：

   def _start(self) -> None:
        assert self._task is None  # 断言任务对象为空
        self._task = asyncio.create_task(self._coro)  # 创建一个异步任务
        self._task.add_done_callback(self._done_callback)  # 为异步任务添加完成回调函数
        self._started.set_result(None)  # 设置开端示例的未来为完成状态

在上面的代码中，我们对未来对象进行set_result(None) 的调用，Future此时将 Future 对象的状态变为 FINISHED，并且你可以通过调用 result() 方法来获取结果，结果将为 None。

追踪发现其实我们的self._started这里没什么作用，不要被魅惑了！

异步后台任务实现

既然我们的知道可以实现类似的异步后台，所以我们结合FatsAPI的话也可以实现类似Background Tasks的效果哟，具体的示例代码如下：

from fastapi import FastAPI
import asyncio
import time

from pyee import AsyncIOEventEmitter

app = FastAPI()

# 创建异步事件发射器
emitter = AsyncIOEventEmitter()
# 一个异步函数，模拟耗时操作

async def callback2(arg1, arg2):
    print("事件2", arg1, arg2)
    await asyncio.sleep(10)
    print("事件2回调处理完成")

# 绑定异步事件和回调函数

emitter.on('event2', callback2)
# 一个异步的 FastAPI 路径操作
@app.get("/")
async def async_endpoint():
    # 调用异步函数
    emitter.emit('event2', '我是', '谁')
    return {"message": "ok"}

# 运行 FastAPI 应用程序
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8200)

在上面的代码，我们发射了事件后，不会阻塞当前API接口的响应请求，也可以立即的返回到我们相应报文。不过需要注意，此类的情况仅仅合适于我们的异步协程的情况，如果你需要是那种同步的情况下，不阻塞当前组线程的，可以考虑基于线程的并发后台异步任务，也就是参考我们的pyee库那种实现的方式。说白了就是把任务异步丢给一个后台的线程池！！！

上面这种方式对比FatsAPI的的Background Tasks的效果，主要是区别于，它支持同步或异步，它内部会自己进行转换处理。而我们的aiojobs则仅仅是针对我们的异步任务而言，才适用。

我们还优化改进，后续我们的所有的任务都定义具体的类去处理，比如如下这种方式：

from fastapi import FastAPI, Request
import asyncio
import time

from pyee import AsyncIOEventEmitter

app = FastAPI()



class MyEventEmitter:
    def __init__(self, emitter: AsyncIOEventEmitter):
        self.emitter = emitter
        self.emitter.on('event2', self.on_event2)
        self.emitter.on('event3', self.on_event3)

    def emit(self, event_name: str, *args):
        self.emitter.emit(event_name, *args)

    async def on_event2(self, arg1, arg2):
        print("事件2", arg1, arg2)
        await asyncio.sleep(10)
        print("事件2回调处理完成")

    async def on_event3(self, arg1, arg2):
        print("事件3", arg1, arg2)
        await asyncio.sleep(2)
        print("事件3回调处理完成")

# 创建异步事件发射器
emitter = AsyncIOEventEmitter()
event_emitter = MyEventEmitter(emitter)

@app.get("/")
async def async_endpoint(request: Request):
    # 创建事件发射器实例

    event_emitter.emit('event2', '我是', '事件2')
    event_emitter.emit('event3', '我是', '事件3')
    return {"message": "ok"}

# 运行 FastAPI 应用程序
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8200)

其实aiojobs源码设计也非常的有意思，值得我们学习相关异步的时候是值得借鉴参考的。深究源码其实感觉其实没其他可以深究的。

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822