第 18 章 · 观察者（Observer）

一句话：一件事发生时，发生的一方只发一条通知；关心这件事的各方自己登记、自己处理，发生的一方不认识它们。 解决的痛：一件事之后要做的事全写在一个函数里，每加一件、撤一件都要改它；其中一件抛异常，整个函数失败，调用方一重试，前面做完的事又做一遍。

18.1 一个场景

你在一家电商公司负责支付回调。用户付完钱，支付平台会调我们的一个接口，告诉我们“订单 P1001 付款成功”，这叫支付回调（payment callback）。我们处理完要回 HTTP 200；平台收到 500 或等太久没响应，就认为通知没送到，再调一次。各家的重试规则不一样，本章简化成“没收到 200 就立即重试，最多 3 次”。

支付成功之后要做四件事，分属四个模块，由不同的小组维护：

库存 Inventory：扣减库存。本章只有一种商品，开始时 10 件。
积分 Points：按实付金额给用户加积分，1 元 1 分。
短信 Sms：给用户发一条“支付成功”。
埋点 Analytics：向数据平台上报一条“已支付”，运营看转化率用。

本章用几个只打印的小类模拟这四个模块；handle_callback 模拟 Web 框架，抛异常就回 500；platform_notify 模拟支付平台的重试。

18.2 先按直觉写

支付成功的处理函数 on_paid 依次调四个模块，全部做完回 200：

from dataclasses import dataclass


@dataclass
class Order:
    order_id: str
    user_id: str
    amount: int  # 实付金额（元）


class Inventory:  # 库存：只有一种商品，开始时 10 件
    def __init__(self) -> None:
        self.stock = 10

    def deduct(self, order: Order) -> None:
        self.stock -= 1
        print(f"库存：订单 {order.order_id} 扣 1 件，剩 {self.stock} 件")


class Points:  # 积分：1 元 1 分
    def __init__(self) -> None:
        self.balance: dict[str, int] = {}

    def add(self, order: Order) -> None:
        self.balance[order.user_id] = self.balance.get(order.user_id, 0) + order.amount
        print(f"积分：{order.user_id} 加 {order.amount} 分")


class Sms:
    def send(self, order: Order) -> None:
        print(f"短信：订单 {order.order_id} 支付成功")


class Analytics:  # 数据埋点
    def track(self, order: Order) -> None:
        print(f"埋点：paid {order.order_id}")


inventory, points, sms, analytics = Inventory(), Points(), Sms(), Analytics()


def on_paid(order: Order) -> str:  # 支付成功后要做的事，依次调四个模块
    inventory.deduct(order)
    points.add(order)
    sms.send(order)
    analytics.track(order)
    return "200 OK"


def handle_callback(order: Order) -> str:  # 模拟 Web 框架：处理函数抛异常就返回 500
    try:
        return on_paid(order)
    except Exception as e:
        return f"500 {e}"


def platform_notify(order: Order) -> None:  # 模拟支付平台：没收到 200 就重试，最多 3 次
    for attempt in range(1, 4):
        result = handle_callback(order)
        print(f"第 {attempt} 次回调返回 {result}")
        if result == "200 OK":
            return


if __name__ == "__main__":
    platform_notify(Order("P1001", "u42", 199))
    print(f"u42 积分余额：{points.balance['u42']}")

# 输出：
# 库存：订单 P1001 扣 1 件，剩 9 件
# 积分：u42 加 199 分
# 短信：订单 P1001 支付成功
# 埋点：paid P1001
# 第 1 次回调返回 200 OK
# u42 积分余额：199

TS 版结构一样：

class Order {
  constructor(
    readonly orderId: string,
    readonly userId: string,
    readonly amount: number, // 实付金额（元）
  ) {}
}

class Inventory { // 库存：只有一种商品，开始时 10 件
  stock = 10;

  deduct(order: Order): void {
    this.stock -= 1;
    console.log(`库存：订单 ${order.orderId} 扣 1 件，剩 ${this.stock} 件`);
  }
}

class Points { // 积分：1 元 1 分
  balance = new Map<string, number>();

  add(order: Order): void {
    this.balance.set(order.userId, (this.balance.get(order.userId) ?? 0) + order.amount);
    console.log(`积分：${order.userId} 加 ${order.amount} 分`);
  }
}

class Sms {
  send(order: Order): void {
    console.log(`短信：订单 ${order.orderId} 支付成功`);
  }
}

class Analytics { // 数据埋点
  track(order: Order): void {
    console.log(`埋点：paid ${order.orderId}`);
  }
}

const inventory = new Inventory();
const points = new Points();
const sms = new Sms();
const analytics = new Analytics();

function onPaid(order: Order): string { // 支付成功后要做的事，依次调四个模块
  inventory.deduct(order);
  points.add(order);
  sms.send(order);
  analytics.track(order);
  return "200 OK";
}

function handleCallback(order: Order): string { // 模拟 Web 框架：处理函数抛异常就返回 500
  try {
    return onPaid(order);
  } catch (e) {
    return `500 ${(e as Error).message}`;
  }
}

function platformNotify(order: Order): void { // 模拟支付平台：没收到 200 就重试，最多 3 次
  for (let attempt = 1; attempt <= 3; attempt++) {
    const result = handleCallback(order);
    console.log(`第 ${attempt} 次回调返回 ${result}`);
    if (result === "200 OK") return;
  }
}

platformNotify(new Order("P1001", "u42", 199));
console.log(`u42 积分余额：${points.balance.get("u42")}`);

// 输出：
// 库存：订单 P1001 扣 1 件，剩 9 件
// 积分：u42 加 199 分
// 短信：订单 P1001 支付成功
// 埋点：paid P1001
// 第 1 次回调返回 200 OK
// u42 积分余额：199

四件事的顺序在 on_paid 里一眼就能看完，出了问题单步跟进去也好查。要做的事不怎么变的时候，这样写清楚直接，大多数人第一版都这么写。

18.3 需求变了

同一个迭代里来了三件事：财务要求支付成功后开电子发票；营销活动期间每笔支付送一次抽奖资格，活动结束就撤掉；埋点服务最近偶尔超时，产品要求埋点失败不能让支付回调报错。

要改的地方有三处，都在支付回调里：

on_paid 加一行 invoice.issue(order)，再新建一个 Invoice 类。开发票是财务小组的事，却要改支付组的函数，整条回调要重新测一遍。
on_paid 再加一行 lottery.grant(order)。活动结束那天还要再改一次 on_paid、再上线一次；忘了删，活动结束后照样发抽奖资格。
给 analytics.track(order) 包一层 try。只包住它一个：发票、抽奖、短信以后要是也会失败，得各自再包。

发票和抽奖先上线了，③ 还在排期。那一周线上的代码是这样（其余部分和 18.2 相同，Analytics 模拟那一周的情况：第一次调用超时）：

class Analytics:  # 那一周的埋点服务：第一次调用超时
    def __init__(self) -> None:
        self.calls = 0

    def track(self, order: Order) -> None:
        self.calls += 1
        if self.calls == 1:
            raise TimeoutError("埋点服务 3 秒未响应")
        print(f"埋点：paid {order.order_id}")


class Invoice:  # ① 新模块：开电子发票
    def issue(self, order: Order) -> None:
        print(f"发票：订单 {order.order_id} 开票 {order.amount} 元")


class Lottery:  # ② 活动期间的临时模块：抽奖资格
    def grant(self, order: Order) -> None:
        print(f"抽奖：{order.user_id} 获得 1 次抽奖资格")


inventory, points, sms, analytics = Inventory(), Points(), Sms(), Analytics()
invoice, lottery = Invoice(), Lottery()


def on_paid(order: Order) -> str:
    inventory.deduct(order)
    points.add(order)
    sms.send(order)
    analytics.track(order)
    invoice.issue(order)  # ① 加一行
    lottery.grant(order)  # ② 加一行，活动结束还要回来删掉
    return "200 OK"

TS 版改的是同样的地方。把这几段换进 before.py（替换原来的 Analytics、创建实例的那一行和 on_paid），支付平台通知一笔 199 元的订单：

库存：订单 P1001 扣 1 件，剩 9 件
积分：u42 加 199 分
短信：订单 P1001 支付成功
第 1 次回调返回 500 埋点服务 3 秒未响应
库存：订单 P1001 扣 1 件，剩 8 件
积分：u42 加 199 分
短信：订单 P1001 支付成功
埋点：paid P1001
发票：订单 P1001 开票 199 元
抽奖：u42 获得 1 次抽奖资格
第 2 次回调返回 200 OK
u42 积分余额：398

第一次回调走到埋点时抛出超时，异常一路抛出 on_paid，Web 框架回了 500。这时库存已经扣了、积分已经加了、短信已经发了，可支付平台只看到 500，又调了一次。第二次从头再来：只卖出一件却剩 8 件；u42 付了 199 元，积分余额是 398；用户收到两条一样的短信。埋点本来是四件事里最不要紧的一件，它一失败，把整个支付回调拖成了 500。

事故之后赶紧补上 ③：

    try:  # ③ 埋点失败不能让回调报错
        analytics.track(order)
    except Exception as e:
        print(f"[日志] 埋点失败：{e}")

可发票、抽奖、短信都没包，哪天短信网关出故障，同样的事故换一个模块再来一次。

问题的形状是：支付成功后要做的事全写在 on_paid 里，每加一件、撤一件都要改支付回调；其中任何一件抛异常，回调就失败，平台一重试，前面做完的事再做一遍。

直觉写法：三处要改的位置和那次 500

图中 ①–③ 就是上面三处，黄色是新加的两行。红色加粗的是埋点：它的异常沿虚线回到 handle_callback 变成 500，平台一重试，三个红框模块又执行一遍。

18.4 用观察者模式重写

换个思路：on_paid 不再去调谁，只宣布“支付成功了”，做成一个带着订单的事件对象 PaymentSucceeded。关心这件事的模块都实现接口 Subscriber，只有一个方法 handle(event)，各自到事件总线 EventBus 登记。EventBus 有三个方法：subscribe 登记，unsubscribe 退订，publish 把事件依次交给每个订阅者。publish 给每个订阅者单独包了 try：一个抛异常只记一行日志，后面的照常调用。

from dataclasses import dataclass
from typing import Protocol


@dataclass
class Order:
    order_id: str
    user_id: str
    amount: int  # 实付金额（元）


@dataclass(frozen=True)
class PaymentSucceeded:  # 事件：支付成功，带上订单
    order: Order


class Subscriber(Protocol):
    def handle(self, event: PaymentSucceeded) -> None: ...


class EventBus:
    def __init__(self) -> None:
        self._subscribers: list[Subscriber] = []

    def subscribe(self, subscriber: Subscriber) -> None:
        self._subscribers.append(subscriber)

    def unsubscribe(self, subscriber: Subscriber) -> None:
        self._subscribers.remove(subscriber)

    def publish(self, event: PaymentSucceeded) -> None:
        for subscriber in list(self._subscribers):  # 复制一份：处理途中有人退订也不乱
            try:
                subscriber.handle(event)
            except Exception as e:  # 一个订阅者失败只记日志，不影响后面的订阅者和发布者
                print(f"[日志] {type(subscriber).__name__} 处理失败：{e}")


class Inventory:
    def __init__(self) -> None:
        self.stock = 10

    def handle(self, event: PaymentSucceeded) -> None:
        self.stock -= 1
        print(f"库存：订单 {event.order.order_id} 扣 1 件，剩 {self.stock} 件")


class Points:
    def __init__(self) -> None:
        self.balance: dict[str, int] = {}

    def handle(self, event: PaymentSucceeded) -> None:
        o = event.order
        self.balance[o.user_id] = self.balance.get(o.user_id, 0) + o.amount
        print(f"积分：{o.user_id} 加 {o.amount} 分")


class Sms:
    def handle(self, event: PaymentSucceeded) -> None:
        print(f"短信：订单 {event.order.order_id} 支付成功")


class Analytics:  # 和 18.3 一样：第一次调用超时
    def __init__(self) -> None:
        self.calls = 0

    def handle(self, event: PaymentSucceeded) -> None:
        self.calls += 1
        if self.calls == 1:
            raise TimeoutError("埋点服务 3 秒未响应")
        print(f"埋点：paid {event.order.order_id}")


class Invoice:  # 新需求：开电子发票，写一个订阅者
    def handle(self, event: PaymentSucceeded) -> None:
        print(f"发票：订单 {event.order.order_id} 开票 {event.order.amount} 元")


class Lottery:  # 活动期间的临时订阅者
    def handle(self, event: PaymentSucceeded) -> None:
        print(f"抽奖：{event.order.user_id} 获得 1 次抽奖资格")


bus = EventBus()
points = Points()
for s in (Inventory(), points, Sms(), Analytics(), Invoice()):  # 订阅者在启动时登记
    bus.subscribe(s)


def on_paid(order: Order) -> str:  # 只发事件，不认识任何订阅者
    bus.publish(PaymentSucceeded(order))
    return "200 OK"


def handle_callback(order: Order) -> str:  # 模拟 Web 框架：处理函数抛异常就返回 500
    try:
        return on_paid(order)
    except Exception as e:
        return f"500 {e}"


def platform_notify(order: Order) -> None:  # 模拟支付平台：没收到 200 就重试，最多 3 次
    for attempt in range(1, 4):
        result = handle_callback(order)
        print(f"第 {attempt} 次回调返回 {result}")
        if result == "200 OK":
            return


if __name__ == "__main__":
    lottery = Lottery()
    bus.subscribe(lottery)  # 活动开始
    platform_notify(Order("P1001", "u42", 199))
    print(f"u42 积分余额：{points.balance['u42']}")
    bus.unsubscribe(lottery)  # 活动结束
    platform_notify(Order("P1002", "u42", 99))
    print(f"u42 积分余额：{points.balance['u42']}")

# 输出：
# 库存：订单 P1001 扣 1 件，剩 9 件
# 积分：u42 加 199 分
# 短信：订单 P1001 支付成功
# [日志] Analytics 处理失败：埋点服务 3 秒未响应
# 发票：订单 P1001 开票 199 元
# 抽奖：u42 获得 1 次抽奖资格
# 第 1 次回调返回 200 OK
# u42 积分余额：199
# 库存：订单 P1002 扣 1 件，剩 8 件
# 积分：u42 加 99 分
# 短信：订单 P1002 支付成功
# 埋点：paid P1002
# 发票：订单 P1002 开票 99 元
# 第 1 次回调返回 200 OK
# u42 积分余额：298

TS 版结构相同，日志里用 constructor.name 取类名：

class Order {
  constructor(
    readonly orderId: string,
    readonly userId: string,
    readonly amount: number, // 实付金额（元）
  ) {}
}

class PaymentSucceeded { // 事件：支付成功，带上订单
  constructor(readonly order: Order) {}
}

interface Subscriber {
  handle(event: PaymentSucceeded): void;
}

class EventBus {
  private subscribers: Subscriber[] = [];

  subscribe(subscriber: Subscriber): void {
    this.subscribers.push(subscriber);
  }

  unsubscribe(subscriber: Subscriber): void {
    this.subscribers = this.subscribers.filter((s) => s !== subscriber);
  }

  publish(event: PaymentSucceeded): void {
    for (const subscriber of [...this.subscribers]) { // 复制一份：处理途中有人退订也不乱
      try {
        subscriber.handle(event);
      } catch (e) { // 一个订阅者失败只记日志，不影响后面的订阅者和发布者
        console.log(`[日志] ${subscriber.constructor.name} 处理失败：${(e as Error).message}`);
      }
    }
  }
}

class Inventory implements Subscriber {
  stock = 10;

  handle(event: PaymentSucceeded): void {
    this.stock -= 1;
    console.log(`库存：订单 ${event.order.orderId} 扣 1 件，剩 ${this.stock} 件`);
  }
}

class Points implements Subscriber {
  balance = new Map<string, number>();

  handle(event: PaymentSucceeded): void {
    const o = event.order;
    this.balance.set(o.userId, (this.balance.get(o.userId) ?? 0) + o.amount);
    console.log(`积分：${o.userId} 加 ${o.amount} 分`);
  }
}

class Sms implements Subscriber {
  handle(event: PaymentSucceeded): void {
    console.log(`短信：订单 ${event.order.orderId} 支付成功`);
  }
}

class Analytics implements Subscriber { // 和 18.3 一样：第一次调用超时
  calls = 0;

  handle(event: PaymentSucceeded): void {
    this.calls += 1;
    if (this.calls === 1) throw new Error("埋点服务 3 秒未响应");
    console.log(`埋点：paid ${event.order.orderId}`);
  }
}

class Invoice implements Subscriber { // 新需求：开电子发票，写一个订阅者
  handle(event: PaymentSucceeded): void {
    console.log(`发票：订单 ${event.order.orderId} 开票 ${event.order.amount} 元`);
  }
}

class Lottery implements Subscriber { // 活动期间的临时订阅者
  handle(event: PaymentSucceeded): void {
    console.log(`抽奖：${event.order.userId} 获得 1 次抽奖资格`);
  }
}

const bus = new EventBus();
const points = new Points();
for (const s of [new Inventory(), points, new Sms(), new Analytics(), new Invoice()]) { // 订阅者在启动时登记
  bus.subscribe(s);
}

function onPaid(order: Order): string { // 只发事件，不认识任何订阅者
  bus.publish(new PaymentSucceeded(order));
  return "200 OK";
}

function handleCallback(order: Order): string { // 模拟 Web 框架：处理函数抛异常就返回 500
  try {
    return onPaid(order);
  } catch (e) {
    return `500 ${(e as Error).message}`;
  }
}

function platformNotify(order: Order): void { // 模拟支付平台：没收到 200 就重试，最多 3 次
  for (let attempt = 1; attempt <= 3; attempt++) {
    const result = handleCallback(order);
    console.log(`第 ${attempt} 次回调返回 ${result}`);
    if (result === "200 OK") return;
  }
}

const lottery = new Lottery();
bus.subscribe(lottery); // 活动开始
platformNotify(new Order("P1001", "u42", 199));
console.log(`u42 积分余额：${points.balance.get("u42")}`);
bus.unsubscribe(lottery); // 活动结束
platformNotify(new Order("P1002", "u42", 99));
console.log(`u42 积分余额：${points.balance.get("u42")}`);

// 输出：
// 库存：订单 P1001 扣 1 件，剩 9 件
// 积分：u42 加 199 分
// 短信：订单 P1001 支付成功
// [日志] Analytics 处理失败：埋点服务 3 秒未响应
// 发票：订单 P1001 开票 199 元
// 抽奖：u42 获得 1 次抽奖资格
// 第 1 次回调返回 200 OK
// u42 积分余额：199
// 库存：订单 P1002 扣 1 件，剩 8 件
// 积分：u42 加 99 分
// 短信：订单 P1002 支付成功
// 埋点：paid P1002
// 发票：订单 P1002 开票 99 元
// 第 1 次回调返回 200 OK
// u42 积分余额：298

观察者版：EventBus 持有一组 Subscriber，on_paid 只调 publish

看黄色的两块：发票和抽奖是新加的订阅者类，其余的框都没动；on_paid 只连着 EventBus。

一次支付成功：publish 依次调用订阅者，埋点抛错被总线接住

看顺序图中间：Analytics 的异常回到 EventBus 就停住了，总线记一行日志，接着调 Invoice 和 Lottery，回调第一次就是 200。

逐条对照 18.3：

① 发票：新写一个 Invoice 订阅者，在启动代码里登记一行，on_paid 一行没动。
② 抽奖：活动开始时 bus.subscribe(lottery)，结束时 bus.unsubscribe(lottery)，不用改代码再上线。真实项目里这两行通常挂在活动配置的起止时间上。输出里 P1002 已经没有抽奖。
③ 埋点：不用单独包 try。publish 对每个订阅者都隔离异常，以后新加的也一样。
事故：埋点照样超时，但只留下一行日志，回调第一次就返回 200，平台不重试，P1001 只加了一次 199 分。

有四件事要说清楚：

登记并没有消失。新订阅者总得有人调 subscribe。想知道“支付成功后到底会发生什么”，不能再只看 on_paid，要把所有 subscribe 都找出来。
吞掉异常也有代价。库存扣减失败，现在同样只是一行日志：钱收了、库存没扣，回调照样 200。所以要分清哪些订阅者“失败了也不要紧”。库存这种必须成功的，要么留在 on_paid 里直接调用、失败就让回调失败，要么失败时记进补偿表由任务重做；日志也要能触发报警。
通知是同步的。publish 在当前线程里一个接一个地调。埋点要是不抛异常、而是卡 3 秒，回调就慢 3 秒；超过平台愿意等的时间，平台照样重试，积分照样加两次。观察者隔离了异常，没有隔离慢。要让慢的订阅者不拖住回调，得把事件放进消息队列，由别的进程异步处理，这是第 30 章事件总线的话题。即使那样，重试仍可能发生（200 回了，网络却断在半路），积分最终要按订单号做幂等（idempotent）：同一个订单只加一次，这不是哪个模式能替你做的。
顺序还是登记顺序。订阅者按 for 里的顺序执行，哪天短信要依赖积分先算完，这个依赖只藏在登记顺序里，见 18.8。

观察者换来的是发生的一方不认识处理的一方，处理的各方互不牵连：加一件事是加一个类、登记一行，撤一件事是退订一行，一个订阅者出错不连累别人。代价是原来一眼能看完的 on_paid，现在是一个接口、一个事件类、一个总线和六个订阅者类；排查“这笔为什么没开发票”，要先查 Invoice 登记了没有，再翻日志。

18.5 模式定义

GoF 对观察者的定义是：定义对象间的一种一对多的依赖关系，当一个对象的状态发生改变时，所有依赖于它的对象都得到通知并被自动更新。

换成本章的话：“一”是支付流程，“多”是库存、积分这些模块；“自动更新”是它们不用来问，支付一成功就被调到。被观察的一方叫主题（Subject），关注它的一方叫观察者（Observer），本章叫订阅者。GoF 原书里，主题就是那个状态会变的对象本身，观察者列表放在它身上；本章把列表单独拿出来放进 EventBus，支付流程持有它、往里发事件。这是很常见的变体，角色还是同一套。

观察者的结构：Subject 持有一组 Observer，notify 时逐个调 update

GoF 图里的 attach、detach、notify 对应本章的 subscribe、unsubscribe、publish，Observer.update() 对应 handle(event)。

通知时带多少数据，GoF 讨论过两种做法。推模型（push model）是把变化的内容放进通知里一起送出去，本章的 PaymentSucceeded 就带着整个订单。拉模型（pull model）是通知只说“我变了”，观察者自己回头向主题取数据：GoF 结构图里 update() 不带参数，ConcreteObserver 再调 subject.getState()，就是拉。推模型省掉一次回查，但事件里改一个字段，所有订阅者都受影响；拉模型各取所需，代价是多一次调用，而且取的时候状态可能又变了。

角色	在本章案例里是谁	职责
主题（Subject）	`EventBus`	保存观察者列表，提供登记、退订和通知
具体主题（ConcreteSubject）	支付流程 `on_paid`	状态发生变化（支付成功）时发出通知，通知里的 `PaymentSucceeded` 带上观察者要的数据
观察者（Observer）	`Subscriber`	接收通知的接口，GoF 叫 `update()`，本章是 `handle(event)`
具体观察者（ConcreteObserver）	`Inventory`、`Points`、`Sms`、`Analytics`、`Invoice`、`Lottery`	收到通知后做自己的那件事

常见场景：

界面数据绑定。数据一变，绑定它的组件各自刷新，Vue 的响应式就是这个思路。
领域事件。“订单已支付”“用户已注册”发出去，积分、通知、推荐各自订阅，就是本章。
日志与监控。业务代码只发“请求完成”，写日志、记指标、报警是三个订阅者。
配置热更新。配置中心的值一变，连接池、限流器各自重新加载，不用重启。
WebSocket 推送。一条行情推给订阅了这个频道的所有连接，连接一断就退订。

18.6 更简洁的写法

本章的订阅者只有一个方法 handle。和第 17 章 17.6 一样，只有一个方法的接口就是一个函数类型：订阅者列表就是 list[Callable[[Order], None]]，登记是 append，退订是 remove，整个“主题”就是这个列表加一个循环：

from collections.abc import Callable
from dataclasses import dataclass


@dataclass
class Order:
    order_id: str
    user_id: str
    amount: int


Handler = Callable[[Order], None]  # 订阅者就是一个"收订单、不返回"的函数
subscribers: list[Handler] = []  # 整个"主题"就是这个列表


def publish(order: Order) -> None:
    for handler in list(subscribers):
        try:
            handler(order)
        except Exception as e:  # 隔离异常仍要自己写
            print(f"[日志] {handler.__name__} 处理失败：{e}")


def deduct_stock(order: Order) -> None:
    print(f"库存：订单 {order.order_id} 扣 1 件")


def add_points(order: Order) -> None:
    print(f"积分：{order.user_id} 加 {order.amount} 分")


def grant_lottery(order: Order) -> None:
    print(f"抽奖：{order.user_id} 获得 1 次抽奖资格")


if __name__ == "__main__":
    subscribers.append(deduct_stock)  # subscribe 就是 append
    subscribers.append(add_points)
    subscribers.append(grant_lottery)  # 活动开始
    publish(Order("P1001", "u42", 199))
    subscribers.remove(grant_lottery)  # 活动结束：unsubscribe 就是 remove
    publish(Order("P1002", "u42", 99))

# 输出：
# 库存：订单 P1001 扣 1 件
# 积分：u42 加 199 分
# 抽奖：u42 获得 1 次抽奖资格
# 库存：订单 P1002 扣 1 件
# 积分：u42 加 99 分

TS 连列表都不用自己写。浏览器和 Node 19 以上都自带 EventTarget 和 CustomEvent，EventTarget 就是一个现成的主题：addEventListener 登记，removeEventListener 退订，dispatchEvent 发布，事件数据放在 CustomEvent 的 detail 里：

interface Order {
  orderId: string;
  userId: string;
  amount: number;
}

const bus = new EventTarget(); // 浏览器和 Node 都自带的"主题"
const orderOf = (e: Event) => (e as CustomEvent<Order>).detail;

function publish(order: Order): void {
  bus.dispatchEvent(new CustomEvent("paid", { detail: order }));
}

const deductStock = (e: Event) => console.log(`库存：订单 ${orderOf(e).orderId} 扣 1 件`);
const addPoints = (e: Event) => console.log(`积分：${orderOf(e).userId} 加 ${orderOf(e).amount} 分`);
const grantLottery = (e: Event) => console.log(`抽奖：${orderOf(e).userId} 获得 1 次抽奖资格`);

bus.addEventListener("paid", deductStock); // subscribe
bus.addEventListener("paid", addPoints);
bus.addEventListener("paid", grantLottery); // 活动开始
publish({ orderId: "P1001", userId: "u42", amount: 199 });
bus.removeEventListener("paid", grantLottery); // 活动结束：unsubscribe
publish({ orderId: "P1002", userId: "u42", amount: 99 });

// 输出：
// 库存：订单 P1001 扣 1 件
// 积分：u42 加 199 分
// 抽奖：u42 获得 1 次抽奖资格
// 库存：订单 P1002 扣 1 件
// 积分：u42 加 99 分

结构还是同一个：主题手里一组订阅者，事情发生时逐个调用。语言替你省掉的是 Subscriber 接口、每个订阅者的类和 EventBus 类。用函数写时有两点要留意：

退订要拿着同一个函数对象。登记时要是写 subscribers.append(lambda o: ...)，手里就没有东西可以拿去退订。removeEventListener 也一样，监听器（listener）要先存进一个变量。
异常隔离各不相同。Python 列表版要自己在 publish 里写 try。EventTarget 自带隔离：一个监听器抛异常，dispatchEvent 照样调后面的监听器，也不把异常抛给发布者。但它也不吞掉：浏览器会把异常报到控制台，Node 则会稍后把它当作未捕获异常（uncaughtException）抛出，没有处理的话进程会退出。

什么时候类仍然值得写：订阅者要带状态（Points 里的余额）；不止一个方法，比如还要声明失败时要不要重试；日志里要看清是谁出错，类名比 <lambda> 好认。

18.7 真实系统里的例子

例一：Django 的信号（Python）

Django 的信号（signals）就是观察者。模型的 save() 写完数据库会发出 post_save 信号，任何应用都可以用 post_save.connect(receiver, sender=Order) 订阅“Order 保存之后”，订单模型不需要知道谁订阅了它。Signal.send 依次调用每个 receiver，返回一个”(receiver, 返回值)“的列表；任何一个 receiver 抛异常，异常会穿过 send 抛出去，后面的 receiver 不再被调用。send_robust 则捕获每个 receiver 的 Exception，把异常对象当作那个 receiver 的返回值放进列表，所有 receiver 都会被调到（较新的版本还会把这些异常写进日志）。真实的 Signal 还有弱引用、dispatch_uid 等，这里省略。下面模拟 connect、disconnect、send 和 send_robust：

from collections.abc import Callable
from typing import Any

Receiver = Callable[..., Any]


class Signal:  # 对应 django.dispatch.Signal；真实版本还有弱引用、dispatch_uid、加锁和缓存
    def __init__(self) -> None:
        self.receivers: list[tuple[Receiver, type | None]] = []

    def connect(self, receiver: Receiver, sender: type | None = None) -> None:
        self.receivers.append((receiver, sender))

    def disconnect(self, receiver: Receiver, sender: type | None = None) -> None:
        self.receivers.remove((receiver, sender))

    def _live(self, sender: type) -> list[Receiver]:  # sender=None 的接收者收所有发送者的信号
        return [r for r, s in self.receivers if s is None or s is sender]

    def send(self, sender: type, **named: Any) -> list[tuple[Receiver, Any]]:  # 依次调用，异常直接抛出去
        return [(r, r(signal=self, sender=sender, **named)) for r in self._live(sender)]

    def send_robust(self, sender: type, **named: Any) -> list[tuple[Receiver, Any]]:  # 异常被捕获，当作结果返回
        responses: list[tuple[Receiver, Any]] = []
        for r in self._live(sender):
            try:
                responses.append((r, r(signal=self, sender=sender, **named)))
            except Exception as err:
                responses.append((r, err))
        return responses


post_save = Signal()  # 对应 django.db.models.signals.post_save


class Order:  # 模拟一个 Django 模型：真实的 Model.save() 写完库就发 post_save
    def __init__(self, order_id: str) -> None:
        self.order_id = order_id

    def save(self) -> None:
        print(f"写库：{self.order_id}")
        post_save.send(sender=Order, instance=self, created=True)


def track(sender: type, instance: Order, **kwargs: Any) -> None:
    raise TimeoutError("埋点服务超时")


def add_points(sender: type, instance: Order, **kwargs: Any) -> str:
    print(f"积分：订单 {instance.order_id} 加积分")
    return "ok"


post_save.connect(track, sender=Order)  # 真实代码常写成 @receiver(post_save, sender=Order)
post_save.connect(add_points, sender=Order)

if __name__ == "__main__":
    order = Order("P1001")
    try:
        order.save()
    except TimeoutError as e:
        print(f"save() 抛出：{e}，add_points 没被调用")
    for receiver, result in post_save.send_robust(sender=Order, instance=order, created=False):
        print(f"send_robust：{receiver.__name__} → {result!r}")

# 输出：
# 写库：P1001
# save() 抛出：埋点服务超时，add_points 没被调用
# 积分：订单 P1001 加积分
# send_robust：track → TimeoutError('埋点服务超时')
# send_robust：add_points → 'ok'

post_save 相当于本章的 EventBus，connect、disconnect、send 就是 subscribe、unsubscribe、publish，sender=Order 则是按发送者过滤：只收 Order 保存的信号。模型的 save() 调的是 send，所以一个 receiver 出错会让 save() 本身抛出，和 18.3 的 on_paid 一样。在信号里做“失败也不要紧”的事，要在 receiver 里自己 try，或者把工作交给任务队列。

例二：Node.js 的 EventEmitter（TypeScript）

Node.js 里很多对象继承自 node:events 的 EventEmitter，比如 HTTP 服务器发 "request"，流发 "data"。on 登记监听器，off 退订，once 只收一次，emit 发布。emit 是同步的：按注册顺序调完所有监听器才返回，返回值表示有没有监听器。某个监听器抛异常，异常直接从 emit 抛给调用者，后面的不再调用，这点和 Django 的 send 一样。"error" 是特殊的事件名：发出 "error" 时如果一个监听器都没有，emit 会把这个错误抛出来，没人捕获的话进程退出，所以 Node 的惯例是给可能出错的 emitter 都挂上 "error" 监听器。下面的精简版 MiniEmitter 和真实的 EventEmitter 跑同一段 demo，再比较输出：

import { EventEmitter } from "node:events";

type Listener = (...args: any[]) => void;

interface Emitter { // node:events 的 EventEmitter 和下面的精简版都满足它
  on(event: string, listener: Listener): this;
  off(event: string, listener: Listener): this;
  once(event: string, listener: Listener): this;
  emit(event: string, ...args: unknown[]): boolean;
}

class MiniEmitter implements Emitter { // 精简版：没有 prependListener、maxListeners、"newListener" 事件等
  private listeners = new Map<string, Listener[]>();

  on(event: string, listener: Listener): this {
    this.listeners.set(event, [...(this.listeners.get(event) ?? []), listener]);
    return this;
  }

  off(event: string, listener: Listener): this { // 真实版本还能用原函数退订 once 注册的监听器，这里省略
    this.listeners.set(event, (this.listeners.get(event) ?? []).filter((l) => l !== listener));
    return this;
  }

  once(event: string, listener: Listener): this { // 包一层：第一次调用时先把自己摘掉
    const wrapper: Listener = (...args) => {
      this.off(event, wrapper);
      listener(...args);
    };
    return this.on(event, wrapper);
  }

  emit(event: string, ...args: unknown[]): boolean {
    const list = this.listeners.get(event) ?? [];
    if (list.length === 0) {
      if (event === "error") throw args[0]; // 没人监听 "error" 就抛出（真实版本对非 Error 值会包一层）
      return false;
    }
    for (const listener of [...list]) listener(...args); // 同步、按注册顺序，异常直接抛给 emit 的调用者
    return true;
  }
}

function demo(bus: Emitter): string[] {
  const log: string[] = [];
  const points = (id: string) => log.push(`积分：${id}`);
  bus.on("paid", (id: string) => log.push(`库存：${id}`));
  bus.on("paid", points);
  bus.once("paid", (id: string) => log.push(`抽奖（只一次）：${id}`));
  log.push(`emit 返回 ${bus.emit("paid", "P1001")}`);
  bus.off("paid", points);
  log.push(`emit 返回 ${bus.emit("paid", "P1002")}`);
  log.push(`没人监听 refund，emit 返回 ${bus.emit("refund", "P1001")}`);
  try {
    bus.emit("error", new Error("埋点服务超时"));
  } catch (e) {
    log.push(`没人监听 error，emit 抛出：${(e as Error).message}`);
  }
  return log;
}

const real = demo(new EventEmitter());
const mini = demo(new MiniEmitter());
real.forEach((line) => console.log(line));
console.log(`精简版和 node:events 输出相同：${JSON.stringify(real) === JSON.stringify(mini)}`);

// 输出：
// 库存：P1001
// 积分：P1001
// 抽奖（只一次）：P1001
// emit 返回 true
// 库存：P1002
// emit 返回 true
// 没人监听 refund，emit 返回 false
// 没人监听 error，emit 抛出：埋点服务超时
// 精简版和 node:events 输出相同：true

EventEmitter 是主题，每个监听器函数是观察者，相当于 18.6 的函数列表再按事件名分了组。once 包一层函数，第一次被调用时先把自己摘掉，再调真正的监听器，18.10 会用到它。

18.8 什么时候别用

只有一个固定的接收者。比如支付成功后只需要把订单改成“已支付”，以后也只有这一件事。直接调 orders.mark_paid(order_id)，一眼看到下一步；改成发事件，要搜遍项目才知道谁在处理，出错时调用栈里只看到 publish 的循环。
事件链太长。订阅者处理完又发出新事件，新事件的订阅者再发事件。比如用户改了订单的收货地址：（1）订单模块发出“订单已修改”；（2）运费订阅者重算运费，发出“金额已变”；（3）优惠订阅者也订阅了“订单已修改”，重算满减后同样发出“金额已变”；（4）发票订阅者订阅“金额已变”，每收到一次就作废重开一张，发出“发票已变”；（5）通知订阅者订阅“发票已变”，每次都给用户发一条短信。用户只改了一个字段，五层传下来，后台作废重开了两张发票，用户收到两条“您的发票已更新”。排查的人要在五个文件之间来回跳，代码里没有任何一处写着“改地址会发短信”。
订阅者之间有顺序依赖，却靠登记顺序保证。短信要写“本单获得 199 积分”，就必须在积分之后执行。这只靠 subscribe 的先后：有人把登记代码按字母重排，或积分改成在另一个文件导入时才登记，短信就先执行，写成“本单获得 0 积分”，不报错。有先后顺序的步骤应该写成顺序调用，或者让积分处理完再发一个“积分已到账”事件，由短信去订阅它。

18.9 容易混淆的模式

模式	意图	区别
观察者（Observer）	一件事发生后通知所有关注者	一对多、单向：主题只管通知，不关心谁收、收了做什么；订阅者之间互不知道
中介者（Mediator）	把一组对象之间的交互集中到一个对象里	多对多、双向协调：中介者认识每个参与者，决定“A 变了之后让 B、C 做什么”。见第 24 章
发布订阅（Publish/Subscribe）	通过消息代理转发消息	中间多了一个消息代理（broker）：双方可以不在同一个进程、不同时在线，消息可以排队、持久化、重投。见第 30 章

两组都能用一个问题当场判断：

观察者还是中介者：问“中间那个对象里有没有业务规则”。本章的 EventBus 只管转发，不知道积分和短信是干什么的，是观察者。如果中间的对象里写着“库存不够就让订单取消、让优惠券退回”，它认识每一方、替它们做决定，那是中介者。
观察者还是发布订阅：问”publish 返回时，订阅者跑完了没有”。本章 publish 返回时，所有订阅者都已执行完。如果 publish 只是把消息交给 Kafka、RabbitMQ 这类代理就返回，订阅者在别的进程里稍后处理，那是发布订阅。两个词常混着叫，区别就在有没有代理。

18.10 练一练

库存预警：写一个 Inventory(sku, stock, threshold)，take(qty) 出库后剩余库存低于阈值，就把商品编号和剩余库存通知给所有订阅者。提供 subscribe、unsubscribe 和一次性订阅 once：用 once 登记的订阅者只收到一次通知，之后自动退订。

写两个订阅者：采购提醒（打印“采购提醒：PHONE-X 只剩 4 件，请补货”）和页面角标（打印“页面角标：PHONE-X 仅剩 4 件”）。库存 10 件、阈值 5：采购提醒用 once 登记，页面角标用 subscribe 登记；依次出库 3、3、1 件，然后退订页面角标，再出库 1 件。Python 和 TS 各写一版。

提示：once 可以照着 18.7 的 MiniEmitter 写，登记一个包装函数，它被调用时先退订自己，再调真正的订阅者。通知时要遍历列表的副本，否则包装函数在循环中途删掉自己，Python 会跳过紧接着的下一个订阅者。

18.11 小结

观察者让发生事情的一方只发通知，关心的各方自己登记、自己处理；加一件事是加一个订阅者，撤一件事是退订一行。
总线逐个隔离订阅者的异常，一个失败不连累别人和发布者；但必须成功的事不该放进“失败也不要紧”的总线。
同步通知隔离不了慢：一个慢订阅者照样拖慢发布者。要异步就得引入队列，见第 30 章；重复处理最终靠幂等解决。
订阅者只有一个方法时，函数列表就够了；TS 有现成的 EventTarget 和 EventEmitter。
和中介者的区别看中间那个对象有没有业务规则；和发布订阅的区别看有没有消息代理。