Advanced Python Mastery 实战 8.1：用生成器定制迭代、实现对象比较与实时监控日志流

• 示例工程
• 教程

本篇文章围绕 David Beazley 的Advanced Python Mastery课程 Exercise 8.1 的官方解答展开，讲解如何借助生成器函数（generator）自定义对象迭代、用迭代快速实现对象相等性比较，以及编写一个类似 Unixtail -f的实时日志流监控器follow()。读完本文，你将掌握__iter__/__eq__的实现套路、生成器的"一次性迭代"特性，以及把文件尾追读封装成可复用通用工具的完整代码，可直接应用于日志监控、数据流管道等实战场景。

本讲修改的文件为structure.py（为Structure基类增加迭代与比较能力），新建的文件为follow.py（流式文件监控生成器）。对应的完整参考实现位于 Solutions/8_1 目录，配套练习原文见 Exercises/ex8_1.md。

(a) 回顾：一个最简单的生成器frange()

在进入正式解答之前，先回顾练习 (a) 建立的直觉：只要想自定义迭代，就应该优先想到生成器函数。生成器写起来很简单——把迭代逻辑写进普通函数，用yield逐个发出值即可。

例如，内置range()不支持小数步长，可以用一个生成器轻松补上：

>>> def frange(start,stop,step): while start < stop: yield start start += step >>> for x in frange(0, 2, 0.25): print(x, end=' ') 0 0.25 0.5 0.75 1.0 1.25 1.5 1.75 >>>

生成器是一次性的迭代对象

需要注意：对同一个生成器对象迭代是一次性操作。第二次for循环不会产生任何输出，因为生成器在第一次迭代结束时已经耗尽：

>>> f = frange(0, 2, 0.25) >>> for x in f: print(x, end=' ') 0 0.25 0.5 0.75 1.0 1.25 1.5 1.75 >>> for x in f: print(x, end=' ') >>>

想重复迭代同一序列，要么重新调用frange()创建新生成器，要么把状态封装进一个带__iter__()的类中——这正是第 (b) 节要用的技巧：

>>> class FRange: def __init__(self, start, stop, step): self.start = start self.stop = stop self.step = step def __iter__(self): n = self.start while n < self.stop: yield n n += self.step >>> f = FRange(0, 2, 0.25) >>> for x in f: print(x, end=' ') 0 0.25 0.5 0.75 1.0 1.25 1.5 1.75 >>> for x in f: print(x, end=' ') 0 0.25 0.5 0.75 1.0 1.25 1.5 1.75 >>>

把__iter__()写成生成器函数，是"让自定义类支持迭代"的最省力方式，因为它自动返回了正确的迭代器协议对象。

(b) 给对象添加迭代：Structure.__iter__()

如果你创建了自定义类，可以通过定义特殊方法__iter__()让它支持迭代。__iter__()返回一个迭代器，最简做法就是像上面的FRange一样把它实现为生成器。

在之前的练习中，你已经定义了一个Structure基类（基于StructureMeta元类，字段通过验证器声明）。现在给它添加__iter__()方法，让实例按字段定义顺序依次产出属性值：

# structure.py ... class Structure(metaclass=StructureMeta): ... def __iter__(self): for name in self._fields: yield getattr(self, name) ...

关键点在于self._fields：它由 Solutions/8_1/structure.py 中的validate_attributes()在类定义时自动收集——遍历类命名空间中的Validator实例并按声明顺序生成cls._fields元组。因此__iter__产出的顺序与字段声明顺序完全一致。

添加完成后，就可以直接迭代实例属性了：

>>> from stock import Stock >>> s = Stock('GOOG', 100, 490.1) >>> for val in s: print(val) GOOG 100 490.1 >>>

其中Stock类（见 Solutions/8_1/stock.py）继承自Structure，用String('name')、PositiveInteger('shares')、PositiveFloat('price')声明字段，因此迭代产出依次为股票名、股数、价格。

(c) 迭代的惊人力量：序列转换、解包与__eq__

Python 在很多你意想不到的地方使用迭代。一旦给Structure加上了__iter__()，立刻就能免费获得一批新能力：转换为序列和元组解包：

>>> s = Stock('GOOG', 100, 490.1) >>> list(s) ['GOOG', 100, 490.1] >>> tuple(s) ('GOOG', 100, 490.1) >>> name, shares, price = s >>> name 'GOOG' >>> shares 100 >>> price 490.1 >>>

list(s)、tuple(s)内部正是通过迭代协议逐个取值；解包name, shares, price = s同样依赖迭代。也就是说，实现一个__iter__就让整个对象序列化了。

顺带实现对象相等性比较

基于tuple(self)，现在可以顺带为Structure添加比较操作符：

# structure.py class Structure(metaclass=StructureMeta): ... def __eq__(self, other): return isinstance(other, type(self)) and tuple(self) == tuple(other) ...

实现要点有两处：

1. isinstance(other, type(self))先确认对方是同类型对象（防止与任意序列误判相等）；
2. tuple(self) == tuple(other)借助第 (b) 节的迭代能力，把两侧字段值转换为元组后逐项比较。

之后就能比较对象了：

>>> a = Stock('GOOG', 100, 490.1) >>> b = Stock('GOOG', 100, 490.1) >>> a == b True >>>

值得强调的是，这一整套能力只需十几行代码：__iter__负责字段值的有序产出，__eq__复用迭代做值比较。单元测试也同步验证了该行为——Solutions/8_1/teststock.py 中的test_eq断言a==b为真，test_repr验证repr(s)输出"Stock('GOOG', 100, 490.1)"。按练习要求重新运行teststock.py的全部用例，此时应全部通过。

(d) 监控流式数据源：follow()生成器

生成器另一个强大用途是简单地产生数据流。本节将编写一个生成器来监视日志文件，实现类似 Unixtail -f的效果。

准备数据源：运行stocksim.py

仓库 Data/stocksim.py 是一个股票市场数据模拟器：它基于 Data/dowstocks.csv 的历史行情，通过MarketSimulator.run()每秒推进模拟时钟（内部以分钟为单位插值），并持续把实时行情记录写入Data/stocklog.csv（见LogPrinter观察者的open(filename,"w")+flush()逻辑）。记录格式为 CSV："名称",价格,"日期","时间",涨跌,开盘,最高,最低,成交量。

在命令行（非 IDLE）进入Data/目录运行它：

% python3 stocksim.py

Windows 用户可以找到stocksim.py双击运行。让该程序在后台持续运行（可运行数小时，无需理会），它就会不断向Data/stocklog.csv追加新行情行。

初版：手工探测文件尾部

先写一个不封装的小程序：打开文件、seek 到末尾、循环探测新数据：

# follow.py import os import time f = open('Data/stocklog.csv') f.seek(0, os.SEEK_END) # Move file pointer 0 bytes from end of file while True: line = f.readline() if line == '': time.sleep(0.1) # Sleep briefly and retry continue fields = line.split(',') name = fields[0].strip('"') price = float(fields[1]) change = float(fields[4]) if change < 0: print('%10s %10.2f %10.2f' % (name, price, change))

运行后你会看到一个实时行情滚动输出（只打印涨跌为负的股票）。这段代码本质上就是tail -f的行为。

关于readline()的说明：这里使用readline()而非通常的for循环读行，是有意为之——它被用来反复探测文件末尾是否追加了新数据：有新数据时返回该行文本，没有新数据时返回空字符串''，据此进入休眠重试。

重构：把数据生产逻辑抽成生成器

观察上面的代码：while循环的前半段在生产数据（读文件），后半段在消费数据（解析并打印）。生成器函数的一大优势是，可以把所有数据生产逻辑搬进一个可复用的函数。

修改代码，让文件读取由生成器函数follow(filename)完成：

# follow.py import os import time def follow(filename): ''' Generator that produces a sequence of lines being written at the end of a file. ''' with open(filename,'r') as f: f.seek(0,os.SEEK_END) while True: line = f.readline() if line == '': time.sleep(0.1) # Sleep briefly to avoid busy wait continue yield line

之后既可以单独使用它：

>>> for line in follow('Data/stocklog.csv'): print(line, end='') ... Should see lines of output produced here ...

也可以把行情打印逻辑改写成依赖follow()的形式：

for line in follow('Data/stocklog.csv'): fields = line.split(',') name = fields[0].strip('"') price = float(fields[1]) change = float(fields[4]) if change < 0: print('%10s %10.2f %10.2f' % (name, price, change))

Solutions/8_1/follow.py 给出了带if __name__ == '__main__':的完整可运行版本，其中f.seek(0, os.SEEK_END)把文件指针移动到文件末尾偏移 0 字节处，只关注此后新追加的内容。

Discussion：为什么需要time.sleep(0.1)

官方解答特意强调：这里的time.sleep()是为了避免忙等待（busy-waiting）。如果不休眠，while True循环会在没有新数据时反复空转readline()，导致 CPU 占用飙到 100%；每轮探测后短暂休眠 0.1 秒，把 CPU 让给其他进程，代价只是最多 100ms 的感知延迟——对日志监控类场景完全可接受。

这一步的深层意义：把迭代模式变成通用积木

官方解答指出："Something very powerful just happened here." 你把一个有趣的迭代模式（读取文件末尾新追加的行）封装成了独立函数。follow()现在是完全通用的工具函数，可放进任何程序使用：监控服务器日志、调试日志，以及任何持续追加的文本数据源。后续练习（如 Exercise 8.2，见 Exercises/ex8_2.md）将在此基础上把follow()的输出接入csv.reader()与Ticker.from_row()，构建数据流管道——这正是"生成器 = 可堆叠的乐高积木"这一心法的起点。

小结与本讲源码索引

Exercise 8.1 的核心收获可归纳为三点：

1. 用生成器定制迭代：__iter__写成生成器函数，即可让任意类支持迭代，并连带获得list()、tuple()、解包等序列化能力；
2. 迭代驱动比较：__eq__借助tuple(self)实现逐字段相等性判断，代码极简且语义清晰；
3. 生成器封装数据流：follow()把"文件尾追读"封装成可复用工具，用time.sleep(0.1)避免忙等待，是后续数据流管道的基础。

本讲全部参考实现与验证材料：

• Solutions/8_1/follow.py：follow()生成器完整实现与使用示例；
• Solutions/8_1/structure.py：Structure基类（含__iter__、__eq__、_fields收集逻辑）；
• Solutions/8_1/stock.py：Stock验证器字段声明示例；
• Solutions/8_1/teststock.py：覆盖迭代、相等性与校验的单元测试；
• Data/stocksim.py：行情数据模拟器，实时向Data/stocklog.csv追加数据；
• Exercises/ex8_1.md：本练习的完整题目原文（含frange与FRange起步示例）。

• 示例工程
• 教程