Python-collections.abc 基础知识

第一章：核心概念与模块定位

Q1：collections.abc 是什么？它和 collections 有什么区别？

A： collections.abc 是 Python 标准库中定义**抽象基类（Abstract Base Classes, ABC）**的子模块，而 collections 主要提供具体的数据结构（如 Counter、deque、OrderedDict）。

模块	内容	示例
collections	具体数据结构	Counter, deque, defaultdict
collections.abc	抽象接口定义	Iterable, Callable, Iterator

注意： Python 3.3+ 中，ABC 从 collections 移到了 collections.abc，直接从 collections 导入 ABC 在 3.10+ 会触发 DeprecationWarning。

Q2：什么是抽象基类（ABC）？为什么要用它？

A： ABC 定义了一组接口规范（必须实现的方法），本身不能被实例化，但可以用来：

1. 类型检查：用 isinstance(obj, SomeABC) 判断对象是否具备某类行为
2. 强制约束：通过继承 ABC，子类必须实现指定方法，否则实例化时报错

from collections.abc import Iterator

# isinstance 检查：判断对象是否为迭代器
gen = (x for x in range(3))
print(isinstance(gen, Iterator))  # True

# 直接实例化 ABC 会报错
# iter = Iterator()  # TypeError: Can't instantiate abstract class Iterator

Q3：collections.abc 和 typing 模块是什么关系？

A： typing 模块中的很多类型是静态类型检查专用（如 typing.Callable、typing.Iterator），而 collections.abc 中的类是运行时真实存在的类，可用于 isinstance() 检查。

from collections.abc import Callable as CallableABC
from typing import Callable as CallableType

def func(): pass

# 运行时检查只能用 collections.abc
print(isinstance(func, CallableABC))   # True
# print(isinstance(func, CallableType))  # ❌ typing.Callable 不能用于 isinstance

---

第二章：Awaitable —— 异步编程基石

Q4：Awaitable 是什么？哪些对象属于 Awaitable？

A： Awaitable 是所有可以被 await 关键字等待的对象的抽象基类。主要包括：

• coroutine（协程函数调用结果）
• Task
• Future

import asyncio
from collections.abc import Awaitable

async def coro():
    return "hello"

task = asyncio.create_task(coro())
future = asyncio.Future()

print(isinstance(coro(), Awaitable))   # True
print(isinstance(task, Awaitable))    # True
print(isinstance(future, Awaitable))  # True

# 普通函数不是 Awaitable
def normal(): pass
print(isinstance(normal, Awaitable))  # False

Q5：Awaitable 和 Coroutine 有什么区别？

A： Coroutine 继承自 Awaitable，是更具体的抽象基类。

抽象基类	含义	实例
Awaitable	可被 await	coroutine, Task, Future
Coroutine	原生协程对象	async def 函数的返回值

from collections.abc import Awaitable, Coroutine

async def my_coro():
    return 1

c = my_coro()
print(isinstance(c, Coroutine))   # True
print(isinstance(c, Awaitable))   # True（Coroutine 继承自 Awaitable）

Q6：如何自定义一个 Awaitable 对象？

A： 实现 __await__ 魔术方法，返回一个迭代器。

import asyncio
from collections.abc import Awaitable

class MyAwaitable:
    def __await__(self):
        # __await__ 必须返回一个迭代器
        async def _impl():
            await asyncio.sleep(0.1)
            return "自定义 Awaitable 完成"
        return _impl().__await__()

async def main():
    obj = MyAwaitable()
    print(isinstance(obj, Awaitable))  # True
    result = await obj
    print(result)

asyncio.run(main())

---

第三章：Callable —— 可调用对象

Q7：Callable 在 collections.abc 中代表什么？

A： Callable 是所有**可以被调用（使用 () 执行）**的对象的抽象基类。函数、类、实现了 __call__ 方法的对象都属于 Callable。

from collections.abc import Callable

# 函数
def func(): pass
print(isinstance(func, Callable))  # True

# 类（类本身可被调用生成实例）
print(isinstance(int, Callable))   # True

# 实现了 __call__ 的实例
class Adder:
    def __call__(self, x, y):
        return x + y

adder = Adder()
print(isinstance(adder, Callable))  # True
print(adder(2, 3))                  # 5

Q8：Callable 和 callable() 内置函数有什么区别？

A： callable(obj) 是内置函数，返回布尔值判断对象是否可调用；isinstance(obj, Callable) 是类型检查，两者结果通常一致，但 isinstance 更常用于类型系统上下文。

from collections.abc import Callable

def test(): pass

print(callable(test))              # True
print(isinstance(test, Callable))  # True

# lambda 也是 Callable
lam = lambda x: x * 2
print(isinstance(lam, Callable))   # True

Q9：如何检查 Callable 的参数和返回值类型？

A： collections.abc.Callable 不支持参数/返回值粒度检查，需配合 typing.Callable 做静态类型注解。

from collections.abc import Callable
from typing import Callable as TypeCallable

# 运行时：只能判断是否可调用
def handler(data: str) -> int:
    return len(data)

print(isinstance(handler, Callable))  # True

# 静态类型：标注参数和返回值类型
processor: TypeCallable[[str], int] = handler

---

第四章：Iterator —— 迭代器

Q10：Iterator 和 Iterable 有什么区别？

A： 这是最容易混淆的一对概念：

概念	抽象基类	必须实现	特征
Iterable（可迭代对象）	Iterable	__iter__()	能被 for 循环遍历
Iterator（迭代器）	Iterator	__iter__() + __next__()	能逐个产出元素，且只能遍历一次

关系： Iterator 继承自 Iterable。迭代器一定是可迭代的，但可迭代对象不一定是迭代器。

from collections.abc import Iterable, Iterator

# 列表是可迭代的，但不是迭代器
lst = [1, 2, 3]
print(isinstance(lst, Iterable))  # True
print(isinstance(lst, Iterator))  # False

# iter() 返回的是迭代器
it = iter(lst)
print(isinstance(it, Iterator))   # True
print(isinstance(it, Iterable))   # True（迭代器继承自 Iterable）

Q11：为什么迭代器只能遍历一次？

A： 迭代器内部维护一个游标/状态，每次调用 __next__() 就前进一次，耗尽后抛出 StopIteration，不会自动重置。

from collections.abc import Iterator

it = iter([1, 2, 3])
print(isinstance(it, Iterator))  # True

print(next(it))  # 1
print(next(it))  # 2
print(next(it))  # 3
# print(next(it))  # StopIteration

# 再次遍历需要重新创建迭代器
it2 = iter([1, 2, 3])

Q12：如何自定义一个迭代器？

A： 继承 Iterator，实现 __iter__()（通常返回 self）和 __next__()。

from collections.abc import Iterator

class CountDown(Iterator):
    def __init__(self, start):
        self.start = start
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.start <= 0:
            raise StopIteration
        self.start -= 1
        return self.start + 1

# 使用
cd = CountDown(3)
print(isinstance(cd, Iterator))  # True

for num in cd:
    print(num)  # 3, 2, 1

Q13：Generator 和 Iterator 是什么关系？

A： Generator 继承自 Iterator。生成器是一种特殊的迭代器，自动实现了 __iter__ 和 __next__，还能通过 send() 接收外部数据。

from collections.abc import Generator, Iterator

def my_gen():
    yield 1
    yield 2

g = my_gen()
print(isinstance(g, Generator))   # True
print(isinstance(g, Iterator))    # True（Generator 继承自 Iterator）

---

第五章：Collection —— 容器家族

Q14：Collection 是什么？它包含哪些子类？

A： Collection 是容器的抽象基类，继承自 Sized（有长度）、Iterable（可迭代）、Container（支持 in 操作）。它是很多具体容器的共同父类。

继承链：

Collection
├── Sequence（有序，支持索引） → list, tuple, str
├── Set（无序不重复）        → set, frozenset  
└── Mapping（键值对）         → dict

from collections.abc import Collection, Sequence, Set, Mapping

print(isinstance([1, 2], Collection))  # True
print(isinstance({1, 2}, Collection))  # True
print(isinstance({"a": 1}, Collection))  # True
print(isinstance((1, 2), Sequence))    # True
print(isinstance({"a": 1}, Mapping))   # True

Q15：Collection 和 Container 有什么区别？

A： Container 只要求支持 in 运算符（实现 __contains__），而 Collection 在此基础上还要求可迭代和可求长度。

from collections.abc import Container, Collection, Sized, Iterable

class OnlyContains(Container):
    def __contains__(self, item):
        return item == "special"

obj = OnlyContains()
print("special" in obj)           # True
print(isinstance(obj, Container)) # True
print(isinstance(obj, Collection)) # False（缺少 __iter__ 和 __len__）

Q16：Sequence 和 Mapping 的核心区别是什么？

A：

特性	Sequence	Mapping
访问方式	整数索引 seq[0]	键索引 map["key"]
有序性	有序（有确定顺序）	3.7+ 字典有序，但 Mapping 不保证
方法要求	__getitem__, __len__	__getitem__, __iter__, __len__
典型成员	list, tuple, str, range	dict, defaultdict, Counter

from collections.abc import Sequence, Mapping

# Sequence 支持整数索引和切片
seq = [10, 20, 30]
print(isinstance(seq, Sequence))  # True

# Mapping 支持键访问和 keys()/values()/items()
mp = {"x": 1, "y": 2}
print(isinstance(mp, Mapping))    # True
print(list(mp.keys()))            # ['x', 'y']

---

第六章：综合对比与实战

Q17：常用 ABC 对照速查表

抽象基类	继承自	必须实现	典型实例	用途
Awaitable	—	__await__	coroutine, Task	异步等待
Coroutine	Awaitable	__await__, send, throw	async def 返回值	原生协程
Callable	—	__call__	函数、类、可调用实例	回调/处理器
Iterable	—	__iter__	list, set, dict	for 循环
Iterator	Iterable	__iter__, __next__	iter(), 生成器	逐个消费
Generator	Iterator	__iter__, __next__, send	yield 函数	生成数据流
Collection	Sized, Iterable, Container	__contains__, __iter__, __len__	list, set, dict	通用容器
Sequence	Collection, Reversible	__getitem__, __len__	list, tuple, str	有序索引
Mapping	Collection	__getitem__, __iter__, __len__	dict	键值存储
Set	Collection	__contains__, __iter__, __len__	set	集合运算

Q18：什么时候应该用 isinstance(x, SomeABC) 而不是 isinstance(x, list)？

A： 当需要判断的是行为/能力而非具体类型时，使用 ABC 更符合"鸭子类型"精神。

from collections.abc import Sequence, Mapping

def process(data):
    # 好的实践：只关心行为，不关心具体类型
    if isinstance(data, Mapping):
        return {k: v * 2 for k, v in data.items()}
    elif isinstance(data, Sequence):
        return [x * 2 for x in data]
    else:
        raise TypeError("需要 Sequence 或 Mapping")

# 同时支持 list, tuple, dict, 甚至自定义类
print(process([1, 2, 3]))           # [2, 4, 6]
print(process((1, 2, 3)))            # [2, 4, 6]
print(process({"a": 1, "b": 2}))    # {'a': 2, 'b': 4}

Q19：collections.abc 和 typing.Protocol 有什么区别？如何选择？

A：

特性	collections.abc	typing.Protocol
检查方式	运行时 isinstance + 注册	静态类型检查（mypy/pyright）
继承要求	需要显式继承或注册	结构化子类型（无需显式继承）
运行时可用	✅ 完全支持	⚠️ runtime_checkable 后才支持 isinstance
适用场景	运行时类型判断	静态类型系统、接口解耦

from typing import Protocol, runtime_checkable
from collections.abc import Callable

# Protocol 方式（静态类型友好）
@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> None: ...

# ABC 方式（运行时友好）
from collections.abc import ABC
class DrawableABC(ABC):
    @abstractmethod
    def draw(self) -> None: ...

# 选择建议：
# - 需要运行时 isinstance 检查 → collections.abc
# - 纯静态类型，解耦依赖 → typing.Protocol

Q20：一个完整的综合运用示例

A： 下面展示如何结合 Awaitable、Callable、Iterator、Collection 写一个类型安全的异步批处理工具：

import asyncio
from collections.abc import Awaitable, Callable, Iterator, Collection

class AsyncBatchProcessor:
    """
    异步批量处理器：接受一个 Collection 的数据，
    用 Callable 处理器逐个处理（支持异步处理器）
    """
    def __init__(
        self,
        handler: Callable[[any], Awaitable[any] | any],
        batch_size: int = 3
    ):
        self.handler = handler
        self.batch_size = batch_size
    
    async def _run_single(self, item: any) -> any:
        """判断处理器是同步还是异步，统一按异步执行"""
        result = self.handler(item)
        if isinstance(result, Awaitable):
            return await result
        return result
    
    async def process(self, data: Collection) -> Iterator:
        """批量异步处理，返回结果迭代器"""
        if not isinstance(data, Collection):
            raise TypeError(f"data 必须是 Collection，得到 {type(data)}")
        
        results = []
        # 分批处理
        batch = []
        for item in data:
            batch.append(item)
            if len(batch) >= self.batch_size:
                # 并发执行一批
                batch_results = await asyncio.gather(
                    *[self._run_single(i) for i in batch]
                )
                results.extend(batch_results)
                batch = []
        
        # 处理剩余
        if batch:
            results.extend(
                await asyncio.gather(*[self._run_single(i) for i in batch])
            )
        
        return iter(results)  # 返回 Iterator

# ========== 测试 ==========

async def async_handler(x: int) -> str:
    await asyncio.sleep(0.01)
    return f"async:{x}"

def sync_handler(x: int) -> str:
    return f"sync:{x}"

async def main():
    data = [1, 2, 3, 4, 5]
    
    # 测试异步处理器
    processor1 = AsyncBatchProcessor(async_handler, batch_size=2)
    result_iter = await processor1.process(data)
    print("异步处理器结果:", list(result_iter))
    
    # 测试同步处理器
    processor2 = AsyncBatchProcessor(sync_handler, batch_size=2)
    result_iter = await processor2.process(data)
    print("同步处理器结果:", list(result_iter))

asyncio.run(main())

---

附录：快速导入模板

from collections.abc import (
    # 异步
    Awaitable, Coroutine, AsyncIterable, AsyncIterator,
    # 可调用
    Callable,
    # 迭代
    Iterable, Iterator, Generator,
    # 容器
    Container, Collection, 
    Sequence, MutableSequence,
    Mapping, MutableMapping,
    Set, MutableSet,
    # 其他
    Sized, Hashable, Reversible
)

---

学习建议： collections.abc 的核心价值在于用接口而非实现编程。在写类型提示或运行时校验时，优先使用 Sequence 而非 list，使用 Mapping 而非 dict，这样代码会更具扩展性和鲁棒性。