适用读者：用 Python 写后端、写着写着开始琢磨”服务之间的关系该怎么组织”的开发者。本篇聊聊 Canary Framework 是怎么从一个小想法长出来的。

起点：FastAPI 很好，但我还想要一层

最开始我用的是 FastAPI。它设计得很好，也很克制：只做它该做的事。

但用久了，拿它和 Java 的 Spring 生态一对比，会感觉 Python 这边并没有一个统一的框架。FastAPI 有依赖注入，但没有 ORM 的部分；每个项目都要自己把各种库拼起来。

于是我就想，给自己设计一个小脚手架。

很多复杂业务，需要一层业务层来缓冲接口层和服务层之间的依赖关系。在 FastAPI 里做这件事是可以的，但需要预先做一个良好的设计——什么放在哪、谁依赖谁、谁先初始化。这对我来说是一个心智负担，至少我是这样感觉的。

而且在用 AI 写代码的时候，这种”靠约定维持”的结构很容易被 AI 的一些小想法带偏：今天在这里多建一个全局对象，明天在那里多加一段初始化，结构就慢慢散了。

每个服务都有自己的生命

所以我开始想服务之间的关系。它其实和微服务的概念差不多：每个服务都有自己的生命——会初始化，会启动，最后会停止。

1
2
3
4
5
6
7
8
9
10
11
12
from canary_framework import Canary, init, start, stop


class Database(Canary):
    @init
    def prepare(self) -> None: ...

    @start
    async def connect(self) -> None: ...

    @stop
    async def close(self) -> None: ...

这样一来，只要我知道一个服务的这三个 API——init()、start()、stop()——我就可以管理这个服务了，不需要知道它内部做了什么。

依赖：一个美妙的递归分治

接下来的问题是，服务之间会有依赖。我不可能在一个服务里做太多的工作，所以会让一个服务依赖另一个服务：

1
2
3
4
5
6
from canary_framework import dep


class UserService(Canary):
    database = dep(Database)
    cache = dep(Cache)

dep() 声明了依赖，同时 self.database 对类型检查器来说就是 Database，IDE 可以直接补全。

有了依赖，就需要在当前服务启动之前，先启动它的所有依赖。而启动它的依赖，又会重复同样的语义——先启动依赖的依赖，再启动它自己。

**这是一个美妙的递归分治。**我只需要对最上层的服务说一句”启动”，整棵树就会按顺序起来：

1
2
async with UserService() as service:
    ...

顺序、循环依赖与并发：一张图

递归分治有三个问题要处理：循环依赖、启动顺序、并发。

我的做法是在启动时维护一张依赖子图：从要启动的服务出发，沿着 dep() 把它依赖的所有服务收集起来。拿到这张图之后：

  • 循环依赖在建图的时候就能发现，在任何服务真正运行之前就报错；
  • 拓扑排序给出了顺序，自然就可以做到”依赖先启动，自己先回收”；
  • 互不依赖的服务可以同时启动，每个服务只需要等自己的依赖完成。

停止是启动的镜像：先停自己，再尝试停依赖。一个依赖可能被好几个服务共用，所以只有在没有人还在用它的时候才会真的停下来。

阶段：不引入状态机

加入并发之后，图中节点的运行顺序已经不可预知了。同一个服务可能被好几条依赖路径同时走到，如果不加处理，它的生命周期就可能被重复执行。

这里我没有引入状态机这种比较重的概念，而是用了一个简单的阶段：每个阶段声明自己的前置阶段。因为生命周期是固定的，所以：

  • 前置阶段没执行，当前阶段就不会执行——比如没有 init 就调用 start，会直接报错，而不是悄悄跳过；
  • 已经执行过的阶段会被跳过，同一个服务的同一个阶段只会运行一次。

init、start、stop 本身就是三个阶段。这个 API 我也开放出来了，大家可以定义自己的阶段，挂在某个生命周期之后：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from canary_framework import Canary, Phase, enter, init, leave

rollback = Phase("rollback")
migrate = Phase("migrate", after=init, leave=rollback)   # 在 init 之后；离开时运行 rollback


class Schema(Canary):
    @migrate
    async def apply(self) -> None: ...

    @rollback
    async def revert(self) -> None: ...


schema = Schema()
await schema.init()
await enter(schema, migrate)    # 运行 @migrate，依赖在前
...
await leave(schema, migrate)    # 运行 @rollback

失败处理：最谨慎的部分

失败处理是一个程序中最谨慎、最重要的部分，对一个框架来说尤其如此。我要考虑清楚：哪些错误应该暴露给使用者，哪些是框架内部可以捕捉处理的，以及出错之后资源怎么回收。

这里做了几个我觉得挺有意思的机制：

  • **失败的服务自己收拾。**一个服务在启动时抛出异常，会立刻执行它自己的 stop，把已经拿到的资源释放掉。
  • **依赖不足就不启动。**依赖它的服务不会再启动，并且会释放它们已经占用的依赖。同时在启动的其他服务不会被中途打断，执行完之后，如果没人用了，也会被回收。所以 start() 抛出异常的时候，它启动的一切都已经收拾干净了，重试只需要再调用一次。
  • **回收不会被打断。**如果启动过程中被取消（比如超时、Ctrl-C），回收仍然会完整执行完，再把取消传递出去。
  • **错误如实暴露。**使用者看到的，是自己的钩子抛出的那个异常；同一个失败不会被重复报告，框架内部的异常也不会漏出去。回收过程中如果又出错，会以 note 的形式附在最初的异常上，不会盖住它。

为了验证这些，我用随机化测试生成各种依赖图、各种失败和取消的组合，检查”每个获取的资源都恰好被回收一次”这类性质。

最后

剩下的就是一些整理的工作：测试、性能、文档、示例等等。

1
pip install canary-framework

这个项目里有大量 AI 生成的内容，代码、测试、文档都有。我已经尽可能地抽时间审阅生成的内容，但肯定还有疏漏，欢迎大家指正。我只是一个新手，也感谢 AI 让我的想法能这么快落地。

如果你对它感兴趣，或者觉得哪里设计得不对，欢迎在 Discussions 里告诉我。谢谢大家。