本文介绍如何在 Python 中对两个生成器进行笛卡尔积遍历时，避免因对象复用导致的数据污染问题，通过 list.copy() 实现轻量级隔离，兼顾性能与正确性。
本文介绍如何在 python 中对两个生成器进行笛卡尔积遍历时，避免因对象复用导致的数据污染问题，通过 `list.copy()` 实现轻量级隔离，兼顾性能与正确性。
在使用 itertools.product 或嵌套循环处理多个生成器时，若被处理的对象（如列表）可能被原地修改（例如调用 .pop()、.append() 或直接赋值），就会引发意外的状态污染——前序迭代修改了后续迭代中仍被复用的同一对象，导致结果不可预测。
例如，以下代码看似合理，实则危险：
def xs():
yield [1, 2]
yield [3, 4]
def ys():
yield [5, 6]
yield [7, 8]
def process(x, y):
print(f'process({x}, {y})')
x.pop() # 修改 x 原对象
y.pop() # 修改 y 原对象
# ❌ 错误：product 复用同一组生成器实例
for x, y in product(xs(), ys()):
process(x, y)
输出为：
process([1, 2], [5, 6]) process([1], [7, 8]) process([3, 4], [5]) process([3], [7])
这是因为 product(xs(), ys()) 内部仅各调用一次 xs() 和 ys()，所有 (x, y) 对共享同一组生成器产出的可变对象引用；而嵌套循环 for x in xs(): for y in ys(): 虽能保证每次 y 都是新生成的，但 x 仍被复用（每个 x 在内层循环中被多次传入 process），因此也不满足“完全隔离”需求。
✅ 正确解法：对每个元素做浅拷贝（shallow copy）
对于仅含不可变元素（如 int、str、float）的一维列表，list.copy()（等价于 x[:] 或 list(x)）即可创建独立副本，开销远低于 copy.deepcopy()：
def cartesian_no_reuse(xs, ys, process):
for x in xs():
for y in ys():
process(x.copy(), y.copy()) # ✅ 安全、高效
该方案确保：
- 每次 process 接收的 x 和 y 都是全新对象；
- 不会相互干扰，也不会影响原始生成器逻辑；
- 时间复杂度接近嵌套循环（仅增加两次 O(n) 浅拷贝），实测性能约为 nested_loops 的 3 倍以内（而非 deepcopy 的百倍开销）。
⚠️ 注意事项：
- list.copy() 仅复制第一层引用。若列表中包含嵌套可变对象（如 [ [1, 2], {'a': 3} ]），仍需 deepcopy ——但此时应优先重构数据结构，避免深层可变嵌套。
- 若生成器产出的是自定义类实例，需确认其支持 .copy() 方法或实现 __copy__；否则可用 copy.copy(obj) 替代。
- 对于不可变类型（如 tuple, str, int），无需拷贝——直接传递即可。
? 性能对比（100×100 元素基准测试）
| 方法 | 平均耗时 | 说明 |
|---|---|---|
| using_product | 0.7 ± 0.0 ms | 复用对象，行为错误 |
| nested_loops | 3.3 ± 0.2 ms | x 复用，y 新建 |
| shallow_copy | 9.6 ± 0.8 ms | ✅ x.copy() + y.copy() |
| with_deepcopy | 426.4 ± 24.9 ms | 过重，不必要 |
可见：浅拷贝在保持语义正确性的前提下，性能损失可控，是生产环境首选方案。
✅ 最佳实践总结
- 避免在笛卡尔积中直接传递可变对象，尤其当处理函数存在副作用时；
- 优先使用 x.copy() / y.copy() 替代 deepcopy，适用于一维数值/字符串列表；
- 若需更高抽象，可封装为可复用工具函数：
from typing import Callable, Iterator, Any
def safe_cartesian(
xs: Callable[[], Iterator[Any]],
ys: Callable[[], Iterator[Any]],
process: Callable[[Any, Any], None],
x_copy: Callable[[Any], Any] = lambda x: x.copy() if hasattr(x, 'copy') else x,
y_copy: Callable[[Any], Any] = lambda y: y.copy() if hasattr(y, 'copy') else y,
) -> None:
for x in xs():
for y in ys():
process(x_copy(x), y_copy(y))
这样既保持灵活性，又杜绝了隐式状态泄漏风险。
