---
title: "德州扑克服务端里的装饰器、闭包、生成器、迭代器和上下文管理器"
description: "用一桌德州扑克说明五个 Python 概念各在什么时候用：闭包做超时自动弃牌的回调，装饰器统一检查是否轮到该座位，迭代器和生成器按行动顺序遍历玩家、按街推进一手牌，上下文管理器让一组动作要么全部生效、要么全部撤销。"
pubDate: 2026-09-29
tags: ["Python"]
---

下面用一桌德州扑克做例子，说明闭包、装饰器、迭代器、生成器和上下文管理器各在什么时候用。顺序按依赖关系排：装饰器建立在闭包之上，生成器是写迭代器的简便方式，上下文管理器又可以用生成器来写。先定义用到的对象：

```python
import copy
import random
from contextlib import contextmanager
from dataclasses import dataclass, field
from functools import wraps

FULL_DECK = tuple(rank + suit for rank in "23456789TJQKA" for suit in "shdc")


@dataclass
class Player:
    name: str
    stack: int
    hole_cards: list = field(default_factory=list)
    bet: int = 0
    folded: bool = False


@dataclass
class Table:
    players: list
    deck: list = field(default_factory=list)
    board: list = field(default_factory=list)
    pot: int = 0
    current_seat: int = 0  # 轮到行动的座位


def new_table():
    deck = list(FULL_DECK)
    random.shuffle(deck)
    players = [Player(name, 1000) for name in ("Alice", "Bob", "Carol")]
    return Table(players=players, deck=deck)
```

## 核心区别

| 概念 | 是什么 | 牌桌上的例子 | 什么时候用 |
|---|---|---|---|
| 闭包 | 记住外层函数变量的内层函数 | 超时自动弃牌的回调 | 要把函数连同一点上下文交出去，稍后由别人调用 |
| 装饰器 | 接收一个函数、返回加工后函数的函数 | 检查是否轮到这个座位 | 同一段前后逻辑要套在很多函数上 |
| 迭代器 | 实现了 `__next__` 的对象，`for` 靠它逐个取值 | 按行动顺序遍历玩家 | 让自定义对象支持 `for` |
| 生成器 | 含 `yield` 的函数，调用后得到一个迭代器 | 一手牌按街推进 | 按需逐个产出值，或者把分步骤的流程写成顺序代码 |
| 上下文管理器 | 实现了 `__enter__` 和 `__exit__`，配合 `with` 使用 | 一组动作要么全部生效、要么全部撤销 | 成对出现、必须配套的操作 |

## 闭包：函数带上一点上下文

轮到的玩家 15 秒不操作，就自动弃牌。计时器到点时，会调用我们交给它的一个函数，这个函数得知道是哪张桌、哪个座位：

```python
def make_auto_fold(table, seat):
    def auto_fold():
        table.players[seat].folded = True

    return auto_fold


table = new_table()
auto_fold = make_auto_fold(table, 1)
auto_fold()  # 计时器到点时调用的就是它
print(table.players[1].folded)  # True
```

`make_auto_fold` 已经返回了，`auto_fold` 仍然记得 `table` 和 `seat`，这就是闭包。交给计时器只要一行：`threading.Timer(15, make_auto_fold(table, 1)).start()`。

闭包记住的是变量本身，不是变量当时的值。给每个座位注册计时器时，如果在循环里直接定义函数：

```python
def register_timeouts(table):
    handlers = []
    for seat in range(len(table.players)):
        def auto_fold():
            table.players[seat].folded = True

        handlers.append(auto_fold)
    return handlers


table = new_table()
register_timeouts(table)[0]()  # 本想让 0 号座位弃牌
print([player.folded for player in table.players])  # [False, False, True]
```

三个 `auto_fold` 共用同一个 `seat`，等到被调用时循环早已结束，`seat` 是 2，弃牌的成了 Carol。把循环体换成 `handlers.append(make_auto_fold(table, seat))` 就对了：每调用一次 `make_auto_fold` 都会新建一个作用域，每个 `auto_fold` 记住各自的 `seat`。

闭包只读外层变量时不用声明；要给外层变量重新赋值，比如计数加一，得先写 `nonlocal`，否则 Python 会把它当成内层函数自己的局部变量。

要把一个函数连同少量上下文交出去，稍后由别人调用，比如计时器、回调、排序用的 `key`，又不值得为此写一个类，就用闭包。

## 装饰器：同一段检查套在所有动作上

下注、跟注、弃牌，每个动作开头都要检查是不是轮到这个座位。把检查写成装饰器，动作函数只管自己的逻辑：

```python
class NotYourTurn(Exception):
    pass


def require_turn(action):
    @wraps(action)
    def wrapper(table, seat, *args, **kwargs):
        if seat != table.current_seat:
            raise NotYourTurn(f"还没轮到 {seat} 号座位")
        return action(table, seat, *args, **kwargs)

    return wrapper


@require_turn
def place_bet(table, seat, amount):
    player = table.players[seat]
    player.stack -= amount
    player.bet += amount
    table.pot += amount


@require_turn
def fold(table, seat):
    table.players[seat].folded = True


table = new_table()
place_bet(table, 0, 20)
try:
    fold(table, 2)
except NotYourTurn as error:
    print(error)  # 还没轮到 2 号座位
print(place_bet.__name__)  # place_bet
```

`@require_turn` 写在函数定义上面，等价于定义完之后执行 `place_bet = require_turn(place_bet)`。`wrapper` 是一个闭包，记住了被装饰的 `action`：先检查，再调用原函数。`@wraps(action)` 把原函数的名字和文档复制到 `wrapper` 上，不加的话 `place_bet.__name__` 会变成 `wrapper`，日志和报错里分不清是哪个动作。

装饰器需要参数时，比如 `@retry(times=3)`，再包一层：最外层接收参数，返回真正的装饰器。

适合写成装饰器的，是很多函数都要的前置或后置逻辑：权限、校验、计时、重试、日志。只有一个函数用到的逻辑，直接写在函数里更清楚。

## 迭代器：让 for 按你的规则走

一轮下注从某个座位开始，顺时针问一圈，弃了牌的人跳过。想让 `for` 直接按这个顺序走，就要实现迭代器协议：

```python
class ActionOrder:
    """从 start 号座位开始转一圈，跳过已经弃牌的玩家"""

    def __init__(self, players, start):
        self.players = players
        self.start = start
        self.offset = 0

    def __iter__(self):
        return self

    def __next__(self):
        while self.offset < len(self.players):
            player = self.players[(self.start + self.offset) % len(self.players)]
            self.offset += 1
            if not player.folded:
                return player
        raise StopIteration


table = new_table()
table.players[1].folded = True
order = ActionOrder(table.players, start=2)
print([player.name for player in order])  # ['Carol', 'Alice']
print([player.name for player in order])  # []
```

`for player in order` 先调用 `iter(order)` 拿到迭代器，再反复调用 `next()`，直到抛出 `StopIteration`。`__iter__` 返回迭代器本身，`__next__` 每次给出下一位还没弃牌的玩家。

第二次遍历什么也拿不到：迭代器是一次性的，走到头就停在那里。列表能反复遍历，是因为每次 `iter(列表)` 都返回一个新的迭代器。

需要让自定义对象支持 `for`、`list()`、`sum()` 这类按顺序取值的操作时，才用得上迭代器。实际代码里很少手写 `__next__`，一般用下一节的生成器。

## 生成器：可以暂停的函数

上一节的 `ActionOrder`，用生成器写只要几行：

```python
def action_order(players, start):
    for offset in range(len(players)):
        player = players[(start + offset) % len(players)]
        if not player.folded:
            yield player


table = new_table()
table.players[1].folded = True
names = [player.name for player in action_order(table.players, start=2)]
print(names)  # ['Carol', 'Alice']
```

函数里有 `yield`，调用它时不会立即执行函数体，而是得到一个生成器对象。生成器就是迭代器：每次 `next()` 运行到下一个 `yield` 就暂停，把值交出去，局部变量原样保留，下次从暂停处继续。`offset` 这样的进度不用再手工存进属性。想让一个类能反复遍历，把它的 `__iter__` 写成生成器函数就行，每次 `for` 都会得到一个新的生成器。

暂停还有更大的用处。一手牌是分阶段推进的：发底牌，下注；发翻牌，下注；发转牌，下注；发河牌，下注。用生成器写，发牌流程就是一段从上到下的顺序代码：

```python
def deal_streets(table):
    for player in table.players:
        player.hole_cards = [table.deck.pop(), table.deck.pop()]
    yield "preflop"
    table.board += [table.deck.pop() for _ in range(3)]
    yield "flop"
    table.board.append(table.deck.pop())
    yield "turn"
    table.board.append(table.deck.pop())
    yield "river"


table = new_table()
for street in deal_streets(table):
    print(street, len(table.board))
    # 这条街的下注在这里进行；只剩一人没弃牌时 break，后面的牌不会再发
# preflop 0
# flop 3
# turn 4
# river 5
```

每个 `yield` 是一次暂停：循环体里进行这条街的下注，下注结束，下一次循环才让生成器接着发牌。只剩一个人没弃牌、这手牌提前结束时直接 `break`，剩下的公共牌根本不会发出来。

算底池这种一次性的汇总，用生成器表达式就够：`sum(player.bet for player in table.players)`，值边产生边累加，不用先建一个列表。

需要按顺序逐个产出值时用生成器，比如遍历、分页读取、无限序列；需要把一个分步骤、中途要等待的流程写成顺序代码时也可以。它和迭代器一样只能走一遍。

## 上下文管理器：成对的操作不会漏掉后一半

一次请求里要连续执行几个动作，比如重放玩家断线期间排队的操作，要求全部生效，或者一个都不生效。进入时拍快照，出错时恢复，正好是一对配套操作：

```python
@contextmanager
def all_or_nothing(table):
    snapshot = copy.deepcopy(table)
    try:
        yield table
    except Exception:
        vars(table).update(vars(snapshot))  # 恢复成进入 with 之前的样子
        raise


table = new_table()
try:
    with all_or_nothing(table):
        place_bet(table, 0, 20)
        place_bet(table, 2, 20)  # 还没轮到 2 号，抛出 NotYourTurn
except NotYourTurn:
    pass
print(table.pot, table.players[0].stack)  # 0 1000
```

`with` 进入时运行 `yield` 之前的代码，拍下快照；`with` 块里抛出的异常会在 `yield` 那一行冒出来，`except` 把牌桌恢复成快照，再 `raise` 让调用方知道失败了。第一笔下注已经执行过，也一起撤销了。快照为什么必须用 `deepcopy`，见[德州扑克服务端里的赋值、浅拷贝和深拷贝](/blog/python-copy-in-td/)。

`@contextmanager` 把生成器变成上下文管理器：`yield` 之前相当于 `__enter__`，之后相当于 `__exit__`，`yield` 出去的值就是 `with ... as x` 里的 `x`。要注意，`yield` 之后的代码只在 `with` 块正常结束时才执行，必须执行的清理，比如释放锁，要放进 `finally`。

成对出现、必须配套的操作都适合交给上下文管理器：加锁和解锁、打开和关闭、开始和回滚。标准库的 `threading.Lock` 和 `open()` 返回的文件对象本身就是上下文管理器，写成 `with lock:`，锁一定会被释放。