自定义容器类型（collections.abc）

Python进阶编程专题 · 创建符合Python协议的自定义容器

专题：Python进阶编程系统学习

关键词：Python, 自定义容器, collections.abc, Sequence, Mapping, Set, 容器协议

一、引言：为什么需要自定义容器

Python之所以如此灵活且强大，很大程度上得益于其"协议式"的编程接口。当你编写 for x in obj 时，Python会在幕后查找 obj.__iter__() 方法；当你使用 obj[key] 时，查找的是 obj.__getitem__()。这套协议体系使得自定义类型可以无缝融入Python的语言生态，享有与内置类型相同的语法支持。

在现实开发中，我们经常需要创建行为类似于列表、字典或集合的类，但又带有额外的业务逻辑：比如一个只读的配置映射、一个按需计算元素的惰性列表、一个自动去重且保持插入顺序的集合。直接继承内置类型（list、dict、set）虽然简单，但在类型检查、接口约束和方法一致性方面存在诸多隐患。这正是 collections.abc 模块的价值所在——它为每一种容器概念定义了清晰的抽象基类（Abstract Base Class, ABC），你只需实现几个核心方法，就能获得整套标准接口。

本章将系统讲解 collections.abc 模块中的主要抽象基类及其使用方法，帮助你写出符合Python哲学的自定义容器类型。

核心概念：Python的容器协议是一种"鸭子类型"（Duck Typing）的设计——只要一个类实现了必要的方法，它就被视为该类型的容器。抽象基类在此基础上提供了正式的接口契约，并通过mix-in方法自动推导出其他方法，大幅减少重复代码。

二、collections.abc 抽象基类体系概述

collections.abc 模块定义了一套层次化的抽象基类，涵盖了Python中所有核心容器概念。理解这个体系是掌握自定义容器的第一步。

2.1 核心抽象基类一览

抽象基类	需要实现的方法	自动获得的方法（mix-in）	典型内置示例
Container	__contains__	（无）	list, dict, set
Hashable	__hash__	（无）	str, tuple
Iterable	__iter__	（无）	list, str, 生成器
Iterator	__next__, __iter__	（无）	generator对象
Collection	__contains__, __iter__, __len__	（无）	list, str, set
Sequence	__getitem__, __len__	__contains__, __iter__, __reversed__, index, count	list, tuple, str
MutableSequence	Sequence + __setitem__, __delitem__, insert	append, clear, reverse, extend, pop, remove, __iadd__	list
Set	__contains__, __iter__, __len__, __le__, __ge__, __lt__, __gt__, __eq__, __ne__, __and__, __or__, __sub__, __xor__, isdisjoint	（大部分通过上述运算符推导）	set, frozenset
MutableSet	Set + add, discard	clear, pop, remove, __ior__, __iand__, __isub__, __ixor__	set
Mapping	__getitem__, __len__, __iter__, __contains__, keys, items, values, __eq__, __ne__	get, __contains__（等）	dict
MutableMapping	Mapping + __setitem__, __delitem__	pop, popitem, clear, update, setdefault	dict, defaultdict

要点：抽象基类的设计遵循"最小实现原则"——你只需要实现最核心的几个"抽象方法"，其余大量方法（如 .count()、.index()、.pop() 等）会通过mix-in机制自动推导。这就是继承ABC的巨大优势。

2.2 类层次结构

这些ABC之间存在清晰的继承关系，了解这棵继承树有助于准确理解各容器的能力边界：

Hashable
Container
Iterable
    Iterator
Collection (Container + Iterable + Sized)
    Sequence (Collection + Reversible)
        MutableSequence
    Set (Collection)
        MutableSet
    Mapping (Collection)
        MutableMapping

从继承树可以看到，任何容器归根结底都是 Container（支持 in 运算）、Iterable（可被迭代）和 Sized（可用 len() 测量）的组合。Collection 将这三种能力合并为一，成为大多数具体容器的直接基类。

三、实现不可变容器（Sequence协议）

不可变序列是自定义容器中最常见的起点。你只需要实现两个方法：__getitem__ 和 __len__，即可自动获得迭代、包含检测、反转、.count() 和 .index() 等全套序列能力。

3.1 基础示例：自定义只读列表

假设我们需要一个"只读的整数序列"，其内部数据由某种业务规则生成，不允许外部修改：

from collections.abc import Sequence


class ReadOnlyIntList(Sequence):
    """一个只读的整数列表，数据由内部规则生成。"""

    def __init__(self, start: int, end: int, step: int = 1):
        self._data = list(range(start, end, step))

    def __getitem__(self, index):
        return self._data[index]

    def __len__(self):
        return len(self._data)

    def __repr__(self):
        return f"ReadOnlyIntList({self._data})"


# 使用示例
r = ReadOnlyIntList(0, 10, 2)
print(len(r))       # 5
print(r[1])         # 2
print(r[1:3])       # [2, 4]
print(4 in r)       # True
print(list(r))      # [0, 2, 4, 6, 8]
print(r.index(6))   # 3
print(r.count(2))   # 1

注意：切片操作 r[1:3] 之所以能自动工作，是因为 Sequence 的mix-in机制检测到 __getitem__ 接收了 slice 对象后，会自动遍历并收集结果。如果你的 __getitem__ 没有处理 slice 类型，需要你显式判断并处理。

3.2 处理切片的高级实现

在某些场景下，你可能希望切片返回的不是普通列表，而是保持自定义类型。这时需要显式处理 slice 对象：

from collections.abc import Sequence


class SquareSequence(Sequence):
    """返回自然数平方的惰性序列。"""

    def __init__(self, length: int):
        self._length = length

    def __getitem__(self, index):
        if isinstance(index, slice):
            # 手动处理切片，返回同类型对象
            start, stop, step = index.indices(self._length)
            return SquareSequence(stop - start)
        if not isinstance(index, int):
            raise TypeError(f"索引必须为整数，而非 {type(index).__name__}")
        if index < 0:
            index += self._length
        if index < 0 or index >= self._length:
            raise IndexError(f"索引 {index} 超出范围")
        return index * index

    def __len__(self):
        return self._length

    def __repr__(self):
        return f"SquareSequence(len={self._length})"


sq = SquareSequence(5)
print(sq[0])    # 0
print(sq[3])    # 9
print(sq[1:4])  # SquareSequence(len=3)
print(list(sq)) # [0, 1, 4, 9, 16]

惰性求值的优势：SquareSequence 并不预先存储所有数值，而是在 __getitem__ 被调用时即时计算。对于概念上无限或极大的序列（如斐波那契数列），这种惰性设计能大幅节省内存。结合 __len__ 的实现，使用者可以清晰知道序列的范围。

3.3 自定义序列的注册机制

除了直接继承ABC，你还可以通过"注册"的方式让现有类被认定为某种抽象基类的虚拟子类，而无需修改原类的继承关系：

from collections.abc import Sequence


class MyList:
    """一个普通的类，没有继承 Sequence。"""
    def __init__(self, items):
        self._items = list(items)

    def __getitem__(self, index):
        return self._items[index]

    def __len__(self):
        return len(self._items)


# 注册为 Sequence 的虚拟子类
Sequence.register(MyList)

ml = MyList([10, 20, 30])
print(isinstance(ml, Sequence))  # True
print(issubclass(MyList, Sequence))  # True
print(ml[0])  # 10
print(len(ml))  # 3

注册机制主要用于适配第三方库中你已经无法修改其继承关系的类。但它有一个代价——注册不会为你添加任何mix-in方法，你必须自己实现所有必需的方法。

四、实现可变容器（MutableSequence协议）

可变序列在不可变序列的基础上增加了修改能力。你需要额外实现 __setitem__、__delitem__ 和 insert 三个方法，然后自动获得 append、extend、pop、remove 和 clear 等方法。

4.1 实现自定义可变列表

from collections.abc import MutableSequence


class TypedList(MutableSequence):
    """一个只能存储同一类型元素的可变列表。"""

    def __init__(self, dtype, iterable=None):
        self.dtype = dtype
        self._data = []
        if iterable:
            for item in iterable:
                self._validate(item)
            self._data = list(iterable)

    def _validate(self, value):
        if not isinstance(value, self.dtype):
            raise TypeError(
                f"期望类型 {self.dtype.__name__}, "
                f"收到 {type(value).__name__}"
            )

    def __getitem__(self, index):
        return self._data[index]

    def __setitem__(self, index, value):
        self._validate(value)
        self._data[index] = value

    def __delitem__(self, index):
        del self._data[index]

    def __len__(self):
        return len(self._data)

    def insert(self, index, value):
        self._validate(value)
        self._data.insert(index, value)

    def __repr__(self):
        return f"TypedList[{self.dtype.__name__}]({self._data})"


# 使用示例
tl = TypedList(int, [1, 2, 3])
print(tl)           # TypedList[int]([1, 2, 3])
tl.append(4)        # 自动获得
print(tl)           # TypedList[int]([1, 2, 3, 4])
tl.pop()            # 自动获得
print(tl)           # TypedList[int]([1, 2, 3])

try:
    tl.append("str")  # 类型错误
except TypeError as e:
    print(e)          # 期望类型 int, 收到 str

注意：MutableSequence 的 __iadd__（即 += 运算符）mix-in实现依赖于 extend 方法，而 extend 又依赖于 __setitem__ 和 __len__。如果你的列表逻辑在 __setitem__ 中有严格验证，+= 操作也会自动继承验证逻辑。

4.2 增强功能的实战示例：事件日志列表

下面实现一个带事件通知的可变列表，每次元素变化时自动触发回调函数。这种模式在观察者设计模式、UI数据绑定和日志系统中非常实用：

from collections.abc import MutableSequence
from typing import Callable, Any


class ObservableList(MutableSequence):
    """当元素发生变化时触发回调的可观察列表。"""

    def __init__(self, iterable=None, on_change: Callable = None):
        self._data = list(iterable) if iterable else []
        self._on_change = on_change or (lambda *_: None)

    def _notify(self, action: str, *args):
        self._on_change(action, *args)

    def __getitem__(self, index):
        return self._data[index]

    def __setitem__(self, index, value):
        old = self._data[index]
        self._data[index] = value
        self._notify("set", index, old, value)

    def __delitem__(self, index):
        old = self._data.pop(index)
        self._notify("del", index, old)

    def __len__(self):
        return len(self._data)

    def insert(self, index, value):
        self._data.insert(index, value)
        self._notify("insert", index, value)

    def __repr__(self):
        return f"ObservableList({self._data})"


# 日志回调
def log_change(action, *args):
    print(f"[变更] {action}: {args}")


ol = ObservableList([1, 2, 3], on_change=log_change)
ol.append(4)       # [变更] insert: (3, 4)
ol[0] = 10         # [变更] set: (0, 1, 10)
del ol[1]          # [变更] del: (1, 2)
ol.pop()           # [变更] del: (3, 4)  —— 自动触发！

五、实现自定义映射类型（Mapping / MutableMapping）

映射类型（也就是字典类型）是Python中使用最广泛的数据结构之一。当你需要创建一个行为类似字典但具有特殊规则的类时（如大小写不敏感的键、带过期时间的缓存、只读配置等），继承 Mapping 或 MutableMapping 是最佳选择。

5.1 不可变映射：只读配置字典

from collections.abc import Mapping


class ReadOnlyConfig(Mapping):
    """一个只读的配置字典，创建后不可修改。"""

    def __init__(self, **kwargs):
        self._store = dict(kwargs)

    def __getitem__(self, key):
        return self._store[key]

    def __len__(self):
        return len(self._store)

    def __iter__(self):
        return iter(self._store)

    def __repr__(self):
        keys = ", ".join(self._store)
        return f"ReadOnlyConfig({keys})"


# 自动获得的方法
config = ReadOnlyConfig(host="localhost", port=8080, debug=True)

print(config["host"])         # localhost
print(len(config))            # 3
print(list(config.keys()))    # ['host', 'port', 'debug']
print(config.get("port"))     # 8080
print(config.get("missing", "默认值"))  # 默认值
print("debug" in config)      # True

# config["host"] = "new"  # TypeError: 'ReadOnlyConfig' 不支持赋值

仅需实现三个核心方法，即可获得 .get()、.keys()、.values()、.items()、__contains__（即 in 运算）等全部映射接口。这种设计非常适合配置管理、API响应封装等不可变数据场景。

5.2 可变映射：大小写不敏感字典

这是一个非常经典的实战需求——无论键以何种大小写存入，都能以任何大小写取出：

from collections.abc import MutableMapping


class CaseInsensitiveDict(MutableMapping):
    """键名大小写不敏感的字典。"""

    def __init__(self, **kwargs):
        self._store = {}
        for key, value in kwargs.items():
            self[key] = value  # 使用 __setitem__ 统一处理

    def _normalize(self, key):
        if isinstance(key, str):
            return key.lower()
        return key

    def __getitem__(self, key):
        return self._store[self._normalize(key)][1]

    def __setitem__(self, key, value):
        normalized = self._normalize(key)
        self._store[normalized] = (key, value)  # 保留原始键名

    def __delitem__(self, key):
        del self._store[self._normalize(key)]

    def __iter__(self):
        return (original_key for original_key, _ in self._store.values())

    def __len__(self):
        return len(self._store)

    def __repr__(self):
        items = ", ".join(
            f"{k!r}: {v!r}"
            for k, v in self.items()
        )
        return f"CaseInsensitiveDict({{{items}}})"


# 使用示例
d = CaseInsensitiveDict(Name="Alice", Email="alice@example.com")
print(d["name"])                # Alice
print(d["EMAIL"])               # alice@example.com
d["NAME"] = "Bob"
print(d["Name"])                # Bob
print(list(d.keys()))           # ['Name', 'Email']
print(d.pop("email"))           # alice@example.com

设计要点：我们在 _store 中存储了元组 (original_key, value)，这样迭代时能返回用户最初使用的键名形式而非全小写版本。这是用户体验上的一个精细考量——用户看到的是他们熟悉的原始键名，而非内部转换后的版本。

5.3 实战：带过期时间的缓存映射

import time
from collections.abc import MutableMapping


class ExpiringCache(MutableMapping):
    """带过期时间的缓存映射。写入 data 秒后自动失效。"""

    def __init__(self, ttl: float = 60.0):
        self._store = {}
        self._ttl = ttl  # 生存时间（秒）

    def _is_expired(self, key):
        if key not in self._store:
            return True
        timestamp, _ = self._store[key]
        return (time.monotonic() - timestamp) > self._ttl

    def _purge(self):
        now = time.monotonic()
        expired = [
            key for key, (ts, _) in self._store.items()
            if (now - ts) > self._ttl
        ]
        for key in expired:
            del self._store[key]

    def __getitem__(self, key):
        if self._is_expired(key):
            del self._store[key]
            raise KeyError(f"键 {key!r} 已过期")
        return self._store[key][1]

    def __setitem__(self, key, value):
        self._store[key] = (time.monotonic(), value)

    def __delitem__(self, key):
        del self._store[key]

    def __iter__(self):
        self._purge()
        return iter(self._store)

    def __len__(self):
        self._purge()
        return len(self._store)

    def __repr__(self):
        return f"ExpiringCache(ttl={self._ttl}s, items={len(self)})"


# 使用示例
cache = ExpiringCache(ttl=2.0)
cache["key1"] = "value1"
print(cache["key1"])   # value1
time.sleep(3)
try:
    print(cache["key1"])  # 已过期
except KeyError as e:
    print(e)             # 键 'key1' 已过期

优化提示：上述实现中的 _purge() 在每次迭代或求长度时清理过期条目。在更追求读写性能的场景中，可以改用惰性清除策略——只在访问过期键时删除，避免每次操作都遍历全部数据。你也可以使用 sortedcontainers 或 heapq 来实现基于优先级的过期清理。

六、实现自定义集合类型（Set / MutableSet）

集合类型专注于成员关系测试和去重操作。由于集合的数学性质（可哈希、无序、互异），其抽象基类要求实现的方法比序列更多，但mix-in带来的收益也更大——你只需实现几个核心运算符，整个集合代数（并集、交集、差集、对称差集、子集判定等）就自动完备了。

6.1 不可变集合：类位图集合

下面实现一个"位图集合"，用整数位运算来表示有限整数集合，极其节省内存：

from collections.abc import Set


class BitSet(Set):
    """用位图表示的整数集合（0 到 n-1 范围内的整数）。"""

    def __init__(self, bits: int = 0, size: int = 64):
        self._bits = bits
        self._size = size  # 最大位数

    def __contains__(self, item):
        if not isinstance(item, int) or item < 0 or item >= self._size:
            return False
        return bool(self._bits & (1 << item))

    def __iter__(self):
        for i in range(self._size):
            if self._bits & (1 << i):
                yield i

    def __len__(self):
        return self._bits.bit_count()

    def __repr__(self):
        return f"BitSet({ {i for i in range(self._size) if self._bits & (1 << i)} })"


# 使用示例
bs = BitSet(0b10101, size=8)  # 包含 0, 2, 4
print(0 in bs)   # True
print(1 in bs)   # False
print(3 in bs)   # False
print(4 in bs)   # True
print(len(bs))   # 3

# 集合运算需要额外实现 __le__, __ge__, __lt__, __gt__,
# __eq__, __ne__, __and__, __or__, __sub__, __xor__, isdisjoint
# 此处为了简洁省略部分实现，python 3.9+ 的 Set 需要这些接口

位图的优势：对于大量整数的集合操作（如数千个整数），位图的内存占用仅为每个元素1个比特位，远低于Python原生 set 的开销。结合位运算实现集合代数，速度极快。这种结构在数据库系统、编译器的活跃变量分析和网络包过滤中广泛使用。

6.2 可变集合：有序唯一集合

标准 set 不保证顺序，但在某些场景（如保留首次插入顺序）下我们需要一个既唯一又有序的数据结构：

from collections.abc import MutableSet


class OrderedSet(MutableSet):
    """保持插入顺序的唯一集合。"""

    def __init__(self, iterable=None):
        self._items = []      # 保持顺序
        self._index = {}      # 元素 -> 索引的映射，用于 O(1) 查找

        if iterable:
            for item in iterable:
                self.add(item)

    def __contains__(self, item):
        return item in self._index

    def __iter__(self):
        return iter(self._items)

    def __len__(self):
        return len(self._items)

    def add(self, item):
        if item not in self._index:
            self._index[item] = len(self._items)
            self._items.append(item)

    def discard(self, item):
        if item in self._index:
            idx = self._index.pop(item)
            self._items[idx] = None  # 标记为删除，避免移位开销

    def __repr__(self):
        items = [repr(item) for item in self if item is not None]
        return f"OrderedSet({{{ {', '.join(items)} }}})"


# 使用示例
os = OrderedSet([3, 1, 2, 1, 4, 3])
print(os)                    # OrderedSet({3, 1, 2, 4})
print(list(os))              # [3, 1, 2, 4]
print(2 in os)               # True
os.discard(1)
print(list(os))              # [3, None, 2, 4]
os.add(5)
print(list(os))              # [3, None, 2, 4, 5]

实现注意：discard 方法将元素标记为 None 而非真正移除，是为了避免删除操作导致的 O(n) 后续元素移位。但在迭代时需要过滤 None 条目。如果需要紧凑存储，可以在合适的时机（如空闲时或元素数量超过阈值时）触发热垃圾回收。Python 3.7+ 的 dict 保证插入顺序，也可以直接用 dict 作为底层存储简化实现。

利用Python 3.7+ 的字典有序特性，我们可以大幅简化 OrderedSet：

from collections.abc import MutableSet


class OrderedSetV2(MutableSet):
    """利用 dict 有序特性实现的优雅版本。"""

    def __init__(self, iterable=None):
        self._dict = {}
        if iterable:
            for item in iterable:
                self.add(item)

    def __contains__(self, item):
        return item in self._dict

    def __iter__(self):
        return iter(self._dict)

    def __len__(self):
        return len(self._dict)

    def add(self, item):
        self._dict[item] = None

    def discard(self, item):
        self._dict.pop(item, None)

    def __repr__(self):
        items = ", ".join(repr(k) for k in self._dict)
        return f"OrderedSetV2({{{items}}})"

七、自定义迭代器与可迭代对象

迭代器协议是Python语言层面的核心协议，所有 for 循环、列表推导式和生成器表达式都依赖它。理解并实现 Iterable 和 Iterator 是构建自定义容器的基础。

7.1 Iterable 与 Iterator 的区别

这是Python初学者最容易混淆的概念：

Iterable（可迭代对象）

实现了 __iter__() 方法
返回一个 Iterator 对象
可以被 for 循环消费
可以被多次迭代（每次返回新的迭代器）
例如：list、str、tuple

Iterator（迭代器）

实现了 __next__() 和 __iter__()
__iter__() 返回 self
是有状态的——只能向前遍历一次
每次 next(it) 消费一个元素
例如：generator、enumerate 的返回值

7.2 实现自定义迭代器

from collections.abc import Iterable, Iterator


class CountDown(Iterator):
    """从 N 倒数到 0 的迭代器。"""

    def __init__(self, start: int):
        self.current = start

    def __next__(self):
        if self.current < 0:
            raise StopIteration
        value = self.current
        self.current -= 1
        return value

    def __iter__(self):
        return self


class CountDownSequence(Iterable):
    """支持多次迭代的倒序计数可迭代对象。"""

    def __init__(self, start: int):
        self.start = start

    def __iter__(self):
        return CountDown(self.start)


# 使用示例
cds = CountDownSequence(3)

# 第一次迭代
print("第一次:", list(cds))  # 第一次: [3, 2, 1, 0]

# 第二次迭代——每次返回新迭代器
print("第二次:", list(cds))  # 第二次: [3, 2, 1, 0]

# 直接使用迭代器
cd = CountDown(5)
print(next(cd))  # 5
print(next(cd))  # 4
print(list(cd))  # [3, 2, 1, 0]

设计原则：可迭代对象（Iterable）和迭代器（Iterator）分离是有意为之的设计。可迭代对象代表"数据的容器"，它不维护迭代状态，因此可以被重复遍历。迭代器代表"遍历的过程"，它维护当前位置，因此通常只能遍历一次。将两者分离是编写正确自定义容器的基础。

7.3 用生成器简化迭代器实现

在大多数实际场景中，你不需要手动编写迭代器类。Python的生成器函数提供了更简洁的等价方案：

class CountDownSequenceV2:
    """使用生成器实现的可迭代对象。"""

    def __init__(self, start: int):
        self.start = start

    def __iter__(self):
        # 生成器函数自动实现 __iter__ 和 __next__
        current = self.start
        while current >= 0:
            yield current
            current -= 1


class Fibonacci(Iterable):
    """生成不超过上限的斐波那契数列。"""

    def __init__(self, limit: int):
        self.limit = limit

    def __iter__(self):
        a, b = 0, 1
        while a <= self.limit:
            yield a
            a, b = b, a + b


fib = Fibonacci(100)
print(list(fib))  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89]
print(list(fib))  # [] —— 因为__iter__每次都创建新生成器

生成器进阶：自Python 3.3起，生成器支持 yield from 语法，可以将迭代逻辑委托给子生成器。自Python 3.6起，异步生成器（async for）支持异步迭代。在 __iter__ 中使用 yield 是最推荐的自定义迭代方式——简洁、正确且高效。

八、容器类型的性能考量与最佳实践

自定义容器的性能取决于底层实现的选择和算法复杂度。以下是构建高性能自定义容器时需要考虑的关键因素。

8.1 时间复杂度对比

操作	list	set	dict	自定义实现建议
索引/键访问	O(1)	N/A	O(1)	使用哈希表或数组作为底层
包含检测（in）	O(n)	O(1)	O(1)	保留哈希索引实现 O(1) 查找
追加	O(1) 摊销	O(1) 摊销	O(1) 摊销	用 list 或 dict 做底层存储
中间插入	O(n)	N/A	N/A	考虑改用链表或blist
删除	O(n)	O(1)	O(1)	需要快速删除时避免列表
迭代	O(n)	O(n)	O(n)	所有容器迭代都是 O(n)
排序	O(n log n)	N/A	N/A	对自定义序列实现 __lt__ 后使用 sorted()

8.2 性能最佳实践

实践一：为包含检测维护哈希索引

如果你的自定义序列需要频繁进行 in 检测，单独维护一个 set 作为辅助索引能大幅提升性能：

class FastContainsList(Sequence):
    """维护哈希索引以实现 O(1) 包含检测的列表。"""

    def __init__(self, iterable=None):
        self._items = []
        self._index = set()  # 哈希索引

        if iterable:
            for item in iterable:
                self._items.append(item)
                self._index.add(item)

    def __getitem__(self, index):
        return self._items[index]

    def __len__(self):
        return len(self._items)

    def __contains__(self, item):
        return item in self._index  # O(1) 而非 O(n)

    def append(self, item):
        self._items.append(item)
        self._index.add(item)

    def __repr__(self):
        return f"FastContainsList({self._items})"

实践二：避免在 getitem 中执行昂贵操作

__getitem__ 是序列最频繁调用的方法（每个元素每次迭代都会调用）。如果你的容器在元素访问时有计算逻辑，考虑添加缓存层：

from functools import lru_cache


class CachedComputation(Sequence):
    """缓存计算结果的序列。"""

    def __init__(self, size: int):
        self._size = size

    @lru_cache(maxsize=None)
    def __getitem__(self, index):
        # 假设这是一个非常昂贵的计算
        if isinstance(index, slice):
            return [self[i] for i in range(*index.indices(self._size))]
        if index < 0:
            index += self._size
        if index < 0 or index >= self._size:
            raise IndexError
        # 模拟耗时计算
        import time
        time.sleep(0.01)  # 10ms 每次
        return index * index * index

    def __len__(self):
        return self._size

实践三：善用 slots 节省内存

对于大量实例的自定义容器类，__slots__ 能显著减少每个实例的内存开销：

# 每个实例约 56 字节开销（无 __slots__）
class Node:
    def __init__(self, value):
        self.value = value
        self.next = None

# 每个实例约 8 字节开销（有 __slots__）
class NodeSlots:
    __slots__ = ('value', 'next')
    def __init__(self, value):
        self.value = value
        self.next = None

性能决策指南：

读多写少 → 使用不可变容器（Sequence/Mapping），配合缓存
写多读少 → 使用可变容器（MutableSequence/MutableMapping），注意插入开销
成员检测频繁 → 维护哈希辅助索引，或用 Set/映射作为底层
大量实例 → 使用 __slots__ 和紧凑数据结构（如 array 或 numpy）
并发访问 → 考虑使用 threading.Lock 包装所有修改方法

九、综合实战：构建一个完整的自定义容器

下面将各个知识点融合起来，实现一个"LRU缓存字典"——它继承 MutableMapping，在容量超标时自动淘汰最久未使用的键值对。这是一个兼具算法复杂度、性能考量和实际工程价值的综合案例。

from collections.abc import MutableMapping
from collections import OrderedDict


class LRUCache(MutableMapping):
    """
    最近最少使用（LRU）缓存字典。

    当缓存项数量超过 capacity 时，自动淘汰最近最少使用的条目。
    所有操作（get/set/delete）的平均复杂度为 O(1)。
    """

    def __init__(self, capacity: int = 128):
        if capacity < 1:
            raise ValueError("capacity 必须 >= 1")
        self._capacity = capacity
        self._cache = OrderedDict()
        self._hits = 0
        self._misses = 0

    @property
    def capacity(self):
        return self._capacity

    @property
    def hits(self):
        return self._hits

    @property
    def misses(self):
        return self._misses

    @property
    def hit_rate(self):
        total = self._hits + self._misses
        return self._hits / total if total > 0 else 0.0

    def __getitem__(self, key):
        try:
            value = self._cache[key]
            self._hits += 1
            # 将访问项移到最右边（最近使用）
            self._cache.move_to_end(key)
            return value
        except KeyError:
            self._misses += 1
            raise

    def __setitem__(self, key, value):
        if key in self._cache:
            # 更新已有项，同时移到末尾
            self._cache[key] = value
            self._cache.move_to_end(key)
        else:
            if len(self._cache) >= self._capacity:
                # 淘汰最久未使用的项
                self._cache.popitem(last=False)
            self._cache[key] = value

    def __delitem__(self, key):
        del self._cache[key]

    def __iter__(self):
        return iter(self._cache)

    def __len__(self):
        return len(self._cache)

    def __repr__(self):
        return (
            f"LRUCache(capacity={self._capacity}, "
            f"size={len(self._cache)}, "
            f"hit_rate={self.hit_rate:.1%})"
        )


# 综合演示
cache = LRUCache(capacity=3)

# 写入 4 个键，最早的 'a' 会被淘汰
cache['a'] = 1
cache['b'] = 2
cache['c'] = 3
cache['d'] = 4

print('a' in cache)      # False —— 被淘汰
print('b' in cache)      # True
print(cache['b'])        # 2 —— 访问后 'b' 成为最近使用

# 再写入一个键，此时淘汰最久未使用的 'c'
cache['e'] = 5
print('c' in cache)      # False —— 被淘汰
print('d' in cache)      # True
print('e' in cache)      # True

print(f"命中了 {cache.hits} 次，命中率 {cache.hit_rate:.1%}")

# 支持 dict 风格的全部操作
print(list(cache.keys()))    # ['b', 'd', 'e']
print(cache.get('missing', '默认'))
cache.update(f=6, g=7)
print(len(cache))            # 3 （淘汰了最旧的）
cache.clear()
print(len(cache))            # 0

实现分析：LRU缓存使用 OrderedDict 作为底层存储，它同时提供了 O(1) 的键查找和有序性维护。move_to_end(key) 将最近访问的项移到字典末尾，popitem(last=False) 从字典头部弹出最久未使用的项。这种双重特性使其成为实现LRU缓存的理想选择。

十、ABC的注册机制与运行时类型检查

除了继承和注册，collections.abc 还提供了灵活的虚拟子类机制和类型检查工具。

10.1 显式注册 vs 隐式子类

ABC提供了两种子类认定方式：

显式注册：通过 SomeABC.register(MyClass) 将 MyClass 注册为虚拟子类。isinstance() 和 issubclass() 会返回 True，但不继承任何mix-in方法。
隐式子类：实现了ABC所需的所有抽象方法后，Python可以自动识别为子类。例如，只要实现了 __getitem__ 和 __len__，即使没有显式继承 Sequence，Python在某些版本中也能通过 __subclasshook__ 识别。

from collections.abc import Sequence


class MySeq:
    """实现了 Sequence 协议，但未继承。"""
    def __getitem__(self, index):
        return index * 2

    def __len__(self):
        return 10


# Python 通过 __subclasshook__ 自动判定
ms = MySeq()
print(isinstance(ms, Sequence))  # 取决于 Python 版本和具体实现

# 但更安全的做法是显式注册
Sequence.register(MySeq)
print(isinstance(ms, Sequence))  # True

10.2 自定义 subclasshook

你可以为自己的抽象基类定义 __subclasshook__，实现更灵活的子类判定逻辑：

from abc import ABC, abstractmethod


class MyContainer(ABC):
    """自定义容器抽象基类。"""

    @abstractmethod
    def __len__(self):
        ...

    @abstractmethod
    def __iter__(self):
        ...

    @classmethod
    def __subclasshook__(cls, subclass):
        if cls is MyContainer:
            has_len = hasattr(subclass, '__len__')
            has_iter = hasattr(subclass, '__iter__')
            if has_len and has_iter:
                return True
        return NotImplemented


# 不需要显式继承或注册
class MyList:
    def __len__(self):
        return 3

    def __iter__(self):
        return iter([1, 2, 3])


print(isinstance(MyList(), MyContainer))  # True

慎用 __subclasshook__：虽然 __subclasshook__ 提供了极大的灵活性，但它通过检查方法存在性来判定类型，可能误判。例如，一个对象有 __len__ 和 __iter__ 方法并不一定意味着它是有效的容器。object 类本身就没有这些方法，但某些代理对象可能会有。在公开API中慎用此机制。

十一、protocol（协议）与 structural typing

从Python 3.8开始，typing.Protocol 提供了静态类型检查层面的协议支持（PEP 544），它和 collections.abc 的抽象基类形成了互补关系。

11.1 定义静态协议

from typing import Protocol, TypeVar, List

T = TypeVar('T')

class SupportsAppend(Protocol[T]):
    """静态类型检查用的协议：支持 .append() 方法。"""
    def append(self, item: T) -> None: ...

def add_item(container: SupportsAppend[int], item: int) -> None:
    container.append(item)

# list 符合 SupportsAppend[int] 协议
add_item([1, 2], 3)

# 不符合协议的类型会被 mypy/pyright 等静态检查器报告
# add_item("str", 3)  # 错误！str 没有 append 方法

11.2 abc 与 Protocol 的对比

维度	collections.abc	typing.Protocol
运行时检查	支持 isinstance/issubclass	支持（用 @runtime_checkable 装饰器）
静态类型检查	有限	完整的静态协议支持
mix-in方法	自动提供	不提供
抽象方法	必须显式实现	鸭子类型，有对应方法即满足
继承关系	正式的类继承	结构子类型（structural subtyping）
适用场景	运行时多态和方法自动推导	静态类型检查、泛型约束

最佳实践：在运行时需要实际方法实现和类型检查的场景使用 collections.abc（如实现自定义容器）；在纯静态类型约束场景使用 typing.Protocol（如函数参数的类型标注）。两者可以共存——一个类可以同时继承abc和满足Protocol。

十二、核心要点总结

1. 协议即接口：Python容器协议本质上是"方法契约"。实现 __getitem__ 和 __len__ 就能获得完整的序列行为。

2. 继承ABC vs 直接实现：继承 collections.abc 中的抽象基类，可以自动获得大量mix-in方法，大幅减少代码量；直接实现协议方法更灵活但需要手写全部功能。

3. 不可变优先：优先实现不可变容器（Sequence、Mapping、Set），它们语义简单、线程安全、易于推理。在确需修改时再升级为可变版本。

4. 惰性求值与缓存：对于计算密集型的自定义容器，使用惰性求值（即时计算）和缓存（如 lru_cache）可以同时获得内存效率和重复访问时的性能。

5. 性能意识：了解底层数据结构的算法复杂度，为 __contains__、__getitem__ 等高频操作选择合适的数据结构（哈希表 vs 数组 vs 位图）。

6. 注册机制的取舍：第三方库类的注册提供了向后兼容的灵活性，但不会自动提供mix-in方法。尽量优先使用继承而非注册。

7. 类型安全：结合 typing.Protocol 进行静态类型检查，可以在开发阶段捕获接口不匹配问题，提升代码健壮性。

8. 迭代器分离原则：始终将可迭代对象（Iterable）和迭代器（Iterator）分开——Iterable每次返回新迭代器，Iterator维护遍历状态。使用生成器是编写迭代器最简洁的方式。

十三、进一步思考

"Protocols are not formal interfaces, but they are the backbone of Python's polymorphism." — Luciano Ramalho, Fluent Python

在掌握了自定义容器类型之后，以下方向值得进一步探索：

异步迭代器：实现 __aiter__ 和 __anext__ 创建支持 async for 的异步容器。
上下文管理器作为容器：实现 __enter__ 和 __exit__让容器支持 with 语句（如自动关闭资源）。
Generic 自定义容器：使用 typing.Generic 让你的容器支持泛型类型标注（如 MyList[int]）。
零拷贝视图：实现类似 NumPy 切片视图的容器，在不复制数据的情况下提供不同"视角"的数据访问。
分布式容器：将元素存储在远程服务器或数据库中，透明地实现 __getitem__ 和 __setitem__。
持久化容器：数据自动持久化到磁盘的容器（如 shelve 模块的原理）。

推荐阅读：Fluent Python 第二版第11-13章深入讲解了Python协议和抽象基类；Python官方文档的 collections.abc 模块参考和 PEP 3119（抽象基类引入）、PEP 544（Protocol静态类型支持）是权威的参考资料。