预备条件：Python web 编程知识，asyncio 异步编程知识，设计模式

前言

这两天写一个小项目要用到 websockets 这个库。看了文档以后发现这个库实现了三种接口：asyncio 异步接口；threading 多线程接口，SansIO 接口。

当时我一看有点蒙，SansIO 是个啥，从来没听说过，于是搜索了官方资料。无奈官方文档有点难读，而且中文互联网上也没有相应的介绍。于是我阅读了一些英文资料，结合自己的理解，尝试写一篇文章讲讲这个概念。

1 一句话简述 SansIO

很简单：

SansIO 是一种架构设计模式，为的是在编写 IO 相关的代码时，将业务逻辑层和 IO 层分开。达到方便移植和测试的目的。

从以下两点来理解这句话：

是一种模式：SansIO 不是一个库，而是一种编程方法，类似于设计模式
分层解耦：用分层的方法来解耦，实现一种两层结构。具体的，SansIO 将代码分为业务逻辑层，专门处理协议等逻辑，而 IO 层处理和字节流的通信。就像计算机网络的 OSI 七层结构，从应用程序层到物理层，每一层都各司其职，组合起来保证了应用的正常运行。

如果你做过比较大型的项目，你应该知道解耦的思想在计算机工程中非常重要。即为尽量保证类似功能的代码都写在一起，这样项目维护起来就更加容易，代码也更方便修改和移植。

SansIO 就是解耦哲学下的架构方法。

2 详细讲讲

对于 IO 编程，工程师往往要做以下两件事：

处理协议：例如我们需要发送一条 HTTP 请求，我们要定义请求头，请求体，url 等数据，然后将这些数据用 urlencode 或者 json 编码然后发送。
处理 IO 字节流：一般的 IO 接口，例如 TCP 和 UDP 协议，直接的处理对象都是字节。在将协议编码后，我们利用相应的接口方法，来发送和接收这些字节流。

可能把你说晕了，接下来就举个简单的例子来说明：

2.1 经典的编程模式

在经典的编程流程里，这两件事我们会一起做。例如要发送一条 http 请求，我们会这么写：

class UserServiceClient:
    def get_name(self, user_id: int) -> Dict[str, Any]:
        return requests.get(
            self.base_url,
            "v1/users/{}/name/".format(user_id)
        )

以上的代码可以正确运行，但是有一个问题，就是难以移植和测试。

假如现在架构师出现在我面前，和我说整个项目要改为异步架构（async/await），那么 IO 相关代码必须要全部重写。这个项目也很难测试，我必须用很多 mock 来模拟要访问的每一个 url。

2.2 SansIO 编程模式

如何优化以上流程呢？

我们可以把事情拆分开来做：先定义好协议，再实现 IO。

不妨先定义一个类，把我们要发送请求的内容全部封装起来：

@dataclass
class RequestDefinition:
    method: str
    path: str
    ...

然后再在发送请求的代码中使用封装好的对象：

import requests
class SyncRequestClient:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url

    def request(self, request_definition: RequestDefinition) -> Any:
        # 使用定义好的 RequestDefinition
        if request_definition.method == “GET”:
            return requests.get(
                self.base_url, request_definition.path, ...
            )
        ...

如果要写成异步方法，那么就：

import aiohttp
class AsyncRequestClient:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url
    async def request(
        self, request_definition: RequestDefinition
    ) -> Any:
        async with aiohttp.ClientSession() as session:
            # 使用定义好的 RequestDefinition
            if request_definition.method == “GET”:
                return await session.get(
                    self.base_url, request_definition.path, ...
                )
            ...