状态即类型：一种设计模式Draft

2024年5月28日

编程的核心在于状态。

计算机科学家会告诉你，编程语言只是定义抽象图灵机DFA（或状态机）的一种方式。另一方面，硬件工程师则会告诉你，CPU——这个运行程序的物理实体——本质上就是由晶体管编码而成的状态机。

编程的核心在于状态。

可以这样认为：忽略语法差异，编程语言之间的区别在于它们如何处理状态。我们可以通过考察状态在编译时与运行时分别解决多少，来对语言的状态处理方式进行分类。

最左侧是完全的解释型语言¹，例如Python和Lua，它们直到运行时才知道要执行什么指令。另一侧则是强类型编译语言，如Rust和C++，它们需要在编译前获取关于程序的详尽信息。

无论语言处于谱系的哪个位置，都存在权衡。要求提供大量类型、内存和生命周期信息的严格语言，通常难以构建和迭代，但在运行时非常高效。而另一侧的语言则相反：易于快速编写，但会带来显著的运行时开销。

强类型语言在一个领域具有明显优势：错误管理。解释型语言的哲学往往是先运行再调试，观察是否抛出意外异常。

算法Y：解释型语言调试流程（简化版）

编写代码

运行代码

若无错误，跳至步骤6

处理错误

跳至步骤2

部署

无论你更看重开发时间还是运行时性能，我想我们都认同：程序应当正确且对错误具有鲁棒性。

我们将探讨一种设计模式，它有助于分类并将运行时逻辑转化为编译时（若使用解释型语言，则是代码编写时）信息。实际上，它将创建一个状态管道，利用语言的类型系统来防止非法的对象状态。别担心，静态类型并非必需。

让我们通过一个使用此模式的重构示例来具体了解。

任务

老板告诉我们，必须编写一个库，用于统计任意 URL 指向的任意文件中给定字符的出现次数。他还希望我们的代码用户能够随时下载该文件。

听起来不错。让我们创建一个类来完成这个任务

class FileDownloadCharCounter:
    def __init__(self, url):
		# 将参数保存为实例变量
        self.url = url

    def download(self):
		# 下载内容并保存供后续使用
        self.downloaded_content = requests.get(self.url).text

    def create_index(self):
		# 创建一个字典来存储字符计数
        self.index = {}
        for char in self.downloaded_content:
            self.index[char] = self.index.get(char, 0) + 1

    def get_count(self, target_char):
		# 从字典中获取计数
        return self.index.get(target_char, 0)

在这个练习中，我们假设 requests.get 永远不会失败。让我们测试一下这个类

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
counter.create_index()
target = "a"
n = counter.get_count("a")
print(f"{target} appears {n} times")

a appears 477 times

很好！可以部署了吗？别急……如果你仔细观察，我们引入了不少未捕获的异常。如果用户这样做会发生什么？

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
target = "a"
n = counter.get_count("a")

AttributeError: 'FileDownloadCharCounter' object has no attribute 'index'

糟糕。如果用户忘记创建索引，他们会得到一个 AttributeError，这没什么帮助。这将迫使他们阅读库的源代码才能找出问题所在，用户体验很糟糕。让我们创建一个不言自明的异常，让用户能够捕获并解决：

class IndexNotCreatedException(Exception):
	pass

class FileDownloadCharCounter:
    def __init__(self, url):
		# 其他代码 ...
        self.index = {}

	# 其他方法 ...

    def get_count(self, target_char):
        if len(self.index) == 0:
            raise IndexNotCreatedException
        return self.index[target_char]

现在我们得到了一个友好的

File "example.py", line 31, in get_count
    raise IndexNotCreatedException
__main__.IndexNotCreatedException

我们可以这样处理它

try:
	count = counter.get_count('a')
except IndexNotCreatedException:
	# 恢复处理
	pass

还有其他可能出错的方式吗？有：

counter.create_index()
target = "a"
n = counter.get_count("a")

  File "example.py", line 13, in create_index
    for char in self.downloaded_content:
AttributeError: 'FileDownloadCharCounter' object has no attribute 'downloaded_content'

如果库用户忘记下载文件，它会抛出另一个 AttributeError，这也没什么帮助。让我们处理一下

class FileNotDownloadedException(Exception):
    pass

class FileDownloadCharCounter:
    def __init__(self, url):
		# ...
        self.downloaded = False

    def download(self):
		# ...
        self.downloaded = True

    def create_index(self):
        if not self.downloaded:
            raise FileNotDownloadedException
		# ...

现在，用户可以像上面那样处理这个异常。但还有一个 bug；你能找到吗？假设用户不小心这样做了

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
counter.create_index()
# 其他重要操作...
counter.create_index()
# 更多重要操作...
target = "a"
n = counter.get_count("a")
print(f"{target} appears {n} times")

现在我们得到了两倍的实际计数！

a appears 954 times

而且 没有异常。这意味着这个 bug 没有被算法 Y 捕获。同样，我们来处理它

class IndexNotCreatedException(Exception):
    pass

class FileDownloadCharCounter:
	# 方法...

    def create_index(self):
        if len(self.index) > 0:
            raise IndexAlreadyCreatedException
		# 更多代码...

呼！这似乎是所有与状态相关的错误了。这是我们的最终代码：

class IndexNotCreatedException(Exception):
    pass

class IndexAlreadyCreatedException(Exception):
    pass

class FileNotDownloadedException(Exception):
    pass

class FileDownloadCharCounter:
    def __init__(self, url):
        self.url = url
        self.index = {}
        self.downloaded = False

    def download(self):
        self.downloaded_content = requests.get(self.url).text
        self.downloaded = True

    def create_index(self):
        if not self.downloaded:
            raise FileNotDownloadedException

        if len(self.index) == 0:
            raise IndexAlreadyCreatedException

        for char in self.downloaded_content:
            self.index[char] = self.index.get(char, 0) + 1

    def get_count(self, target_char):
        if len(self.index) == 0:
            raise IndexNotCreatedException

        return self.index.get(target_char, 0)

瞧！我们的面向对象代码……糟透了。对于本应是几行简单的 Python 代码，我们却有了 3 个自定义异常和一堆逻辑，仅仅是为了确保没有出错。

你可以想象，如果跟踪状态的实例变量不是 3 个，而是 30 个，我们可能甚至无法知道或枚举所有非法状态来抛出异常。

不幸的是，许多解释型语言往往使得编写健壮的代码变得繁琐。为了帮助解决这个问题，让我们引入一种新的设计模式，我将其称为类型管道，或者你愿意的话，也可以叫 Typeline。

存在性 $im pl i es$ 状态

通过这个技巧，在静态类型语言中，含有非法状态的代码将无法编译。在动态类型语言中，如果你没有收到 TypeError，那么你将拥有一个保证合法的状态。

其工作原理如下：

将类型 $T$ 与状态 $Y$ 关联起来
保证类型 $T$ 的实例对象 $t$ 的存在意味着我们处于状态 $Y$

就这么简单。现在让我们尝试重构之前的代码。考虑我们程序的有效状态：

状态	已知 URL	文件已下载	文件已索引
1	$c h ec kma r k$
2	$c h ec kma r k$	$c h ec kma r k$
3	$c h ec kma r k$	$c h ec kma r k$	$c h ec kma r k$

很明显，这是一个简单的线性流程。状态 1 需要一个 URL 作为输入，状态 2 需要状态 1，状态 3 需要状态 2。

如果 $T_{n}$ 是对应于状态 $N$ 的类型，那么构造 $T_{n}$ 的实例应该只能通过 $T_n - 1$ 的实例来访问。让我们从状态 1 开始，它只需要一个有效的 URL。

class FileURL: # 即 T_1
	def __init__(self, url):
		# 可选地验证 url
		self.url = url

由于我们只希望 $T_{2}$ 由 $T_{1}$ 的实例构造，让我们在 $T_{1}$ 上添加一个方法，用于构造一个具有有效状态（文件已下载）的 $T_{2}$ 。

class FileURL:
	# 其他方法 ...
	def download_file(self, file_url: FileURL) -> DownloadedFile:
		file_contents = requests.get(file_url).text
		return DownloadedFile(file_contents)

class DownloadedFile:
	def __init__(self, file_contents: str):
		self.contents = file_contents

我们对 $T_{3}$ 重复此过程，它代表一个已索引的文件。

class DownloadedFile:
	# 其他方法 ...
	def index_file(self) -> IndexedFile:
		index = {}
		for char in content:
            index[char] = index.get(char, 0) + 1
		return IndexedFile(index)

class IndexedFile:
	def __init__(self, index: dict[str, int]):
		self.index = index

	def get_count(self, target_char):
		return self.index.get(target_char, 0)

现在，所有这些类都只是为了演示这个想法。实际上，我们并不真的需要让用户访问一个已下载但未索引的文件，也不需要仅仅包装一个字符串的 URL 类。他们只需要能够按需下载文件并获取字符计数。

因此，我们可以移除与那些状态关联的类型，并将逻辑塞进一个转换方法或构造函数中。

class FileURL:
	def __init__(self, url):
		self.url = url

	# 状态 1 -> 3
	def fetch_index(self) -> CharIndex:
		# 状态 1 -> 2
		content = requests.get(url).text
		# 状态 2 -> 3
		return CharIndex(content)

class CharIndex:
	# 状态 2 -> 3
	def __init__(self, content: str):
		index = {}
		for char in content:
            index[char] = index.get(char, 0) + 1
		self.index = index

	def get_count(self, target_char):
		return self.index.get(target_char, 0)

用法：

file = FileURL("https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc")
index = file.fetch_index()
count = index.get_count('a')

你可以看到，有了转换/构造的保证，非法状态不再是一个问题。

然而，有一些注意事项。这种模式只有在以下条件下才有效：

状态在编译/代码编写时是已知的（即第 $n$ 行的对象应该具有状态 $X$ ）
只有少数几个这样的状态，因为你可能需要为每个状态创建一个新类

部分优势

在垃圾回收语言中，一个巨大的优势是内存效率。回到我们的例子，在原始类中，下载的文件与整个对象具有相同的生命周期。这意味着，只要存在对对象的引用，整个站点的字符串就会一直存储在内存中。

counter = FileDownloadCharCounter(...)
counter.download_file()  # 为字符串分配空间
counter.create_index()   # 为索引字典分配空间
counter.get_count('a')

# 所有内容在作用域结束时释放

但我们从任务中知道，索引创建后，就不再需要文件内容了。因为Typeline围绕数据转换构建，生命周期是明确定义的。如果一段数据不再需要，垃圾回收器可以安全地将其销毁。

file = FileURL(...)
index = file.fetch_index() # 字符串被分配并立即释放
count = index.get_count('a') # 索引被分配
# 索引在作用域结束时释放

如果我们下载的文件有数GB大小，或者我们同时抓取数千个网站，这一点尤其有用。

总结

本文描述了一个简单的设计模式，我在编写 streamrip v2 时发现它很有用，它修复了 v1 中极大量可能出现的状态错误，并显著简化了代码库。虽然我之前没有遇到过完全相同的想法，但这绝非原创。Typeline 只是一种面向对象的方式，用于编码带有副作用管理的确定性有限自动机（DFA）。因此，所有关于 DFA 的理论工作在这里同样适用。

如果你在其他地方见过这种模式，请告诉我！

我们称之为解释型的现代语言实际上是通过 JIT（即时）编译进行编译的。我们使用“解释型”这个术语，是指类型错误的程序仍然可以运行。 ↩︎

✦ 本文的构思、研究、撰写和编辑均未使用大语言模型。

状态即类型：一种设计模式Draft

任务

存在性 implies 状态

部分优势

总结

存在性 $im pl i es$ 状态