德州扑克服务端里的装饰器、闭包、生成器、迭代器和上下文管理器

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

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 秒不操作，就自动弃牌。计时器到点时，会调用我们交给它的一个函数，这个函数得知道是哪张桌、哪个座位：

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()。

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

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，又不值得为此写一个类，就用闭包。

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

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

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 直接按这个顺序走，就要实现迭代器协议：

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，用生成器写只要几行：

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 都会得到一个新的生成器。

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

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)，值边产生边累加，不用先建一个列表。

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

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

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

@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，见德州扑克服务端里的赋值、浅拷贝和深拷贝。

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

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