第 12 章 · 代理（Proxy）

一句话：给真实对象配一个接口相同的替身，调用方以为在用真对象，替身决定转不转、什么时候转、转之前先做什么。 解决的痛：权限、懒加载、缓存、审计这些把关的事，要么全塞进真实对象，要么散在每个调用处；新写一个调用处的人只要漏判一次，私密数据就从那里出去了。

12.1 一个场景

你在一家做相册应用的公司写后端。用户的照片原图放在对象存储里，每张几 MB。读图的是一个图片服务 ImageStore，只有一个方法 load(path)：给一个路径，从对象存储拉回原图。本章用一个计数器加模拟耗时（每张 300 毫秒）代替真的网络，拉了几次、花了多久都打印出来。

相册分公开和私密两种。路径以 private/ 开头的是私密相册，第二段是主人的用户名，比如 private/alice/01.jpg 只有 alice 能看；其余路径谁都能看。判断写在用户对象上：user.can_view(path)。

用到图片的页面目前有两个：

• 相册列表页 album_page：一屏展示 50 张卡片，用户点开哪张就大图显示哪张。
• 详情页 open_photo：别人发来的分享链接，直接打开一张。

本章有两个用户：alice 有一个私密相册，bob 看公开的旅行相册，也可能拿到 alice 私密照片的链接。

12.2 先按直觉写

最直接的写法是：列表页把 50 张都拉回来做成卡片，点开时直接显示；每个用到图片的地方先判一下权限。为了省时间，有人在 ImageStore 里加了一个缓存字典，拉过的图不再拉第二次：

from dataclasses import dataclass

FETCH_MS = 300  # 从对象存储拉一张原图的模拟耗时（毫秒）


@dataclass
class Photo:  # 拉回来的原图，真实的有几 MB
    path: str
    size_kb: int

    def show(self) -> str:
        return f"显示 {self.path}（{self.size_kb} KB）"


@dataclass
class User:
    name: str

    def can_view(self, path: str) -> bool:  # private/<主人>/... 只有主人能看，其余公开
        parts = path.split("/")
        return parts[0] != "private" or parts[1] == self.name


class ImageStore:  # 从对象存储拉原图（模拟）；后来有人在里面加了缓存字典
    def __init__(self) -> None:
        self.fetches = 0  # 真正拉取的次数
        self.elapsed_ms = 0
        self.cache: dict[str, Photo] = {}

    def load(self, path: str) -> Photo:
        if path in self.cache:
            return self.cache[path]
        self.fetches += 1
        self.elapsed_ms += FETCH_MS
        photo = Photo(path, 2048)
        self.cache[path] = photo
        return photo


store = ImageStore()


def album_page(user: User, paths: list[str]) -> list[Photo]:  # 相册列表页：每张图一张卡片
    return [store.load(p) for p in paths if user.can_view(p)]  # 调用处自己判权限


def click(card: Photo) -> str:  # 用户点开一张卡片
    return card.show()


def open_photo(user: User, path: str) -> str:  # 详情页：从分享链接直接打开一张
    if not user.can_view(path):  # 这里也要自己判权限
        return f"{user.name} 无权查看 {path}"
    return store.load(path).show()


if __name__ == "__main__":
    alice, bob = User("alice"), User("bob")
    trip = [f"public/trip/{i:02d}.jpg" for i in range(1, 51)]
    cards = album_page(bob, trip)
    print(f"列表页：{len(cards)} 张卡片，拉取 {store.fetches} 次，模拟耗时 {store.elapsed_ms / 1000} 秒")
    print(click(cards[0]))
    print(open_photo(bob, "private/alice/01.jpg"))
    print(open_photo(alice, "private/alice/01.jpg"))
    print(f"一共拉取 {store.fetches} 次")

# 输出：
# 列表页：50 张卡片，拉取 50 次，模拟耗时 15.0 秒
# 显示 public/trip/01.jpg（2048 KB）
# bob 无权查看 private/alice/01.jpg
# 显示 private/alice/01.jpg（2048 KB）
# 一共拉取 51 次

TS 版结构一样，缓存用 Map：

const FETCH_MS = 300; // 从对象存储拉一张原图的模拟耗时（毫秒）

class Photo { // 拉回来的原图，真实的有几 MB
  constructor(readonly path: string, readonly sizeKb: number) {}

  show(): string {
    return `显示 ${this.path}（${this.sizeKb} KB）`;
  }
}

class User {
  constructor(readonly name: string) {}

  canView(path: string): boolean { // private/<主人>/... 只有主人能看，其余公开
    const parts = path.split("/");
    return parts[0] !== "private" || parts[1] === this.name;
  }
}

class ImageStore { // 从对象存储拉原图（模拟）；后来有人在里面加了缓存
  fetches = 0; // 真正拉取的次数
  elapsedMs = 0;
  private readonly cache = new Map<string, Photo>();

  load(path: string): Photo {
    const hit = this.cache.get(path);
    if (hit !== undefined) return hit;
    this.fetches += 1;
    this.elapsedMs += FETCH_MS;
    const photo = new Photo(path, 2048);
    this.cache.set(path, photo);
    return photo;
  }
}

const store = new ImageStore();

function albumPage(user: User, paths: string[]): Photo[] { // 相册列表页：每张图一张卡片
  return paths.filter((p) => user.canView(p)).map((p) => store.load(p)); // 调用处自己判权限
}

function click(card: Photo): string { // 用户点开一张卡片
  return card.show();
}

function openPhoto(user: User, path: string): string { // 详情页：从分享链接直接打开一张
  if (!user.canView(path)) return `${user.name} 无权查看 ${path}`; // 这里也要自己判权限
  return store.load(path).show();
}

const alice = new User("alice");
const bob = new User("bob");
const trip = Array.from({ length: 50 }, (_, i) => `public/trip/${String(i + 1).padStart(2, "0")}.jpg`);
const cards = albumPage(bob, trip);
console.log(`列表页：${cards.length} 张卡片，拉取 ${store.fetches} 次，模拟耗时 ${(store.elapsedMs / 1000).toFixed(1)} 秒`);
console.log(click(cards[0]));
console.log(openPhoto(bob, "private/alice/01.jpg"));
console.log(openPhoto(alice, "private/alice/01.jpg"));
console.log(`一共拉取 ${store.fetches} 次`);

// 输出：
// 列表页：50 张卡片，拉取 50 次，模拟耗时 15.0 秒
// 显示 public/trip/01.jpg（2048 KB）
// bob 无权查看 private/alice/01.jpg
// 显示 private/alice/01.jpg（2048 KB）
// 一共拉取 51 次

两个页面各自判权限，bob 打不开 alice 的私密照片，alice 自己能打开；缓存让同一张图只拉一次。一个相册几十张图、两个页面时，这样写清楚直接，大多数人第一版都这么写。列表页一打开就拉 50 次、模拟耗时 15 秒，相册小的时候没人觉得慢。

12.3 需求变了

产品要上一个新功能“导出相册”，同时来了三条要求：私密相册的图不管从哪个入口读，都必须先鉴权；每次读原图都要记一条审计日志，写清谁读了哪张；列表页不许一打开就拉 50 张原图，用户点开哪张再拉哪张。导出功能交给了刚入职的同事。

要改的地方有四处：

1. ImageStore.load：审计日志要记“谁”，load 只好多一个参数 user；日志得写在查缓存之前，否则命中缓存的那些读取会漏记。拉取、缓存、日志三件事挤在一个方法里，ImageStore 已经不只是“从对象存储拉图”了。签名一变，所有调用 load 的地方都要跟着多传一个 user。
2. album_page：不再调 load，卡片里只剩路径。卡片的类型从 Photo 变成了字符串，点开卡片的 click 不能再直接 card.show()，要改成拿着路径和用户去调 open_photo；以后哪个页面也要展示卡片，都得照这个样子写一遍。
3. 权限判断：album_page 和 open_photo 里各有一句 if user.can_view(path)，新写的导出功能也得记得写这一句。没有任何东西提醒他。
4. 新增 export_album：照着别的页面调 store.load。

改完的样子（只列改动，Photo、User 和 12.2 相同）：

class ImageStore:
    def __init__(self) -> None:
        self.fetches = 0
        self.elapsed_ms = 0
        self.cache: dict[str, Photo] = {}
        self.audit: list[str] = []  # ① 审计日志也放进来

    def load(self, path: str, user: User) -> Photo:  # ① 为了记"谁读的"，多了参数 user
        self.audit.append(f"{user.name} 读取 {path}")  # 必须写在查缓存之前，否则命中缓存的读取会漏记
        if path in self.cache:
            return self.cache[path]
        self.fetches += 1
        self.elapsed_ms += FETCH_MS
        photo = Photo(path, 2048)
        self.cache[path] = photo
        return photo


def album_page(user: User, paths: list[str]) -> list[str]:  # ② 不再拉图，卡片里只剩路径
    return [p for p in paths if user.can_view(p)]


def click(user: User, card: str) -> str:  # ② 卡片从 Photo 变成 str，点开时才拉，还得多传 user
    return open_photo(user, card)


def open_photo(user: User, path: str) -> str:
    if not user.can_view(path):  # ③ 判权限照旧写在调用处
        return f"{user.name} 无权查看 {path}"
    return store.load(path, user).show()  # ① 跟着多传 user


def export_album(user: User, paths: list[str]) -> list[Photo]:  # ④ 新功能，新同事写的
    return [store.load(p, user) for p in paths]  # 传了 user，他以为 store 会查权限

事故出在第 4 处。load 的签名里有 user，新同事很自然地以为传进去的用户会在里面被检查，就没写 can_view；评审时也没人注意。把上面几段换进 before.py，main 里让 bob 打开列表页、点开第一张，再导出 alice 的私密相册：

    alice, bob = User("alice"), User("bob")
    trip = [f"public/trip/{i:02d}.jpg" for i in range(1, 51)]
    cards = album_page(bob, trip)
    print(f"列表页：{len(cards)} 张卡片，拉取 {store.fetches} 次")
    print(click(bob, cards[0]))
    secret = [f"private/alice/{i:02d}.jpg" for i in range(1, 4)]
    exported = export_album(bob, secret)
    print(f"bob 导出了 {len(exported)} 张：{'、'.join(p.path for p in exported)}")
    print(f"审计日志 {len(store.audit)} 条，最后一条：{store.audit[-1]}")

运行结果：

列表页：50 张卡片，拉取 0 次
显示 public/trip/01.jpg（2048 KB）
bob 导出了 3 张：private/alice/01.jpg、private/alice/02.jpg、private/alice/03.jpg
审计日志 4 条，最后一条：bob 读取 private/alice/03.jpg

懒加载做到了，列表页一张没拉；可 bob 把 alice 的三张私密照片导了出来。审计日志倒是忠实地记下了“bob 读取 private/alice/03.jpg”，只是记下来时照片已经出去了。TS 版要改的是同样的四处，事故也一样。

问题的形状是：把关的事（权限、日志、缓存、什么时候拉）要么挤进 ImageStore 让它越长越大，要么散在每个调用处靠人记住；每多一个用到图片的入口，就多一个可能漏判的地方，而漏判的那个入口照样能跑。

图中 ①–④ 就是上面四处。边框加粗的 export_album 通往 User.can_view 的是一条虚线：它直接到了 ImageStore，中间没人拦。

12.4 用代理模式重写

换个思路：页面要的只是“给我一张图”，至于这张图要不要先查权限、记不记日志、是不是早就拉过、该不该现在就拉，都不是页面该操心的。那就让页面只认两个接口：Image（能 show）和 ImageSource（能 load）。真正拉图的 RemoteImageStore 只管拉取；查权限、记日志、缓存各写成一个代理，接口和 RemoteImageStore 一样，手里拿着下一层；列表页的卡片是 LazyImage，第一次 show 时才去 load；它只替一张图占位，所以冒充的是 Image，不是图片服务。

from dataclasses import dataclass
from typing import Protocol

FETCH_MS = 300  # 从对象存储拉一张原图的模拟耗时（毫秒）


class Image(Protocol):  # 页面眼里的"一张图"：能显示就行
    def show(self) -> str: ...


class ImageSource(Protocol):  # 页面眼里的"图片服务"：给路径，拿到一张图
    def load(self, path: str) -> Image: ...


@dataclass
class Photo:  # 拉回来的原图，和 12.2 相同
    path: str
    size_kb: int

    def show(self) -> str:
        return f"显示 {self.path}（{self.size_kb} KB）"


@dataclass
class User:  # 和 12.2 相同
    name: str

    def can_view(self, path: str) -> bool:
        parts = path.split("/")
        return parts[0] != "private" or parts[1] == self.name


class RemoteImageStore:  # 真实对象：只管从对象存储拉图
    def __init__(self) -> None:
        self.fetches = 0
        self.elapsed_ms = 0

    def load(self, path: str) -> Image:
        self.fetches += 1
        self.elapsed_ms += FETCH_MS
        return Photo(path, 2048)


class CachingProxy:  # 缓存代理：拉过的图留在内存
    def __init__(self, inner: ImageSource) -> None:
        self.inner = inner
        self.cache: dict[str, Image] = {}

    def load(self, path: str) -> Image:
        if path not in self.cache:
            self.cache[path] = self.inner.load(path)
        return self.cache[path]


class AuditProxy:  # 记审计日志：谁读了哪张
    def __init__(self, inner: ImageSource, user: User, log: list[str]) -> None:
        self.inner, self.user, self.log = inner, user, log

    def load(self, path: str) -> Image:
        self.log.append(f"{self.user.name} 读取 {path}")
        return self.inner.load(path)


class AccessProxy:  # 保护代理：没权限就不往里转
    def __init__(self, inner: ImageSource, user: User) -> None:
        self.inner, self.user = inner, user

    def load(self, path: str) -> Image:
        if not self.user.can_view(path):
            raise PermissionError(f"{self.user.name} 无权查看 {path}")
        return self.inner.load(path)


class LazyImage:  # 虚代理：替一张图占位，第一次 show 才去拉
    def __init__(self, path: str, source: ImageSource) -> None:
        self.path, self.source = path, source
        self.real: Image | None = None

    def show(self) -> str:
        if self.real is None:
            self.real = self.source.load(self.path)
        return self.real.show()


# ---- 组装处：整个程序只有这里 new RemoteImageStore，页面和导出拿到的都是 ImageSource ----
audit_log: list[str] = []
_remote = RemoteImageStore()
_cache = CachingProxy(_remote)  # 全站共享一份缓存，放在权限检查里面


def image_source_for(user: User) -> ImageSource:  # 顺序：先查权限，再记日志，再查缓存，最后才拉
    return AccessProxy(AuditProxy(_cache, user, audit_log), user)


# ---- 下面是页面和功能，只认 Image 和 ImageSource ----
def album_page(paths: list[str], source: ImageSource) -> list[Image]:  # 列表页：50 张卡片，一张都不拉
    return [LazyImage(p, source) for p in paths]


def click(card: Image) -> str:  # 和 12.2 一样，点开就是 show
    return card.show()


def export_album(paths: list[str], source: ImageSource) -> list[Image]:  # 新同事的写法没变，只是手里的 source 变了
    return [source.load(p) for p in paths]


if __name__ == "__main__":
    alice, bob = User("alice"), User("bob")
    trip = [f"public/trip/{i:02d}.jpg" for i in range(1, 51)]
    cards = album_page(trip, image_source_for(bob))
    print(f"列表页：{len(cards)} 张卡片，拉取 {_remote.fetches} 次")
    print(f"bob 点开第 1 张：{click(cards[0])}，拉取 {_remote.fetches} 次")
    print(f"bob 再点第 1 张：{click(cards[0])}，拉取 {_remote.fetches} 次")
    alice_cards = album_page(trip[:1], image_source_for(alice))
    print(f"alice 点开同一张：{click(alice_cards[0])}，拉取 {_remote.fetches} 次")

    secret = [f"private/alice/{i:02d}.jpg" for i in range(1, 4)]
    try:
        export_album(secret, image_source_for(bob))
    except PermissionError as error:
        print(f"bob 导出 alice 的私密相册：拒绝，{error}")
    print(f"alice 导出自己的私密相册：{len(export_album(secret, image_source_for(alice)))} 张，拉取 {_remote.fetches} 次")
    print(f"审计日志 {len(audit_log)} 条：")
    for line in audit_log:
        print(f"  {line}")

# 输出：
# 列表页：50 张卡片，拉取 0 次
# bob 点开第 1 张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
# bob 再点第 1 张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
# alice 点开同一张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
# bob 导出 alice 的私密相册：拒绝，bob 无权查看 private/alice/01.jpg
# alice 导出自己的私密相册：3 张，拉取 4 次
# 审计日志 5 条：
#   bob 读取 public/trip/01.jpg
#   alice 读取 public/trip/01.jpg
#   alice 读取 private/alice/01.jpg
#   alice 读取 private/alice/02.jpg
#   alice 读取 private/alice/03.jpg

TS 版用构造参数属性持有下一层，LazyImage 用 ??= 在第一次 show 时赋值：

const FETCH_MS = 300; // 从对象存储拉一张原图的模拟耗时（毫秒）

interface Image { // 页面眼里的"一张图"：能显示就行
  show(): string;
}

interface ImageSource { // 页面眼里的"图片服务"：给路径，拿到一张图
  load(path: string): Image;
}

class Photo implements Image { // 拉回来的原图，和 12.2 相同
  constructor(readonly path: string, readonly sizeKb: number) {}

  show(): string {
    return `显示 ${this.path}（${this.sizeKb} KB）`;
  }
}

class User { // 和 12.2 相同
  constructor(readonly name: string) {}

  canView(path: string): boolean {
    const parts = path.split("/");
    return parts[0] !== "private" || parts[1] === this.name;
  }
}

class PermissionError extends Error {}

class RemoteImageStore implements ImageSource { // 真实对象：只管从对象存储拉图
  fetches = 0;
  elapsedMs = 0;

  load(path: string): Image {
    this.fetches += 1;
    this.elapsedMs += FETCH_MS;
    return new Photo(path, 2048);
  }
}

class CachingProxy implements ImageSource { // 缓存代理：拉过的图留在内存
  private readonly cache = new Map<string, Image>();

  constructor(private readonly inner: ImageSource) {}

  load(path: string): Image {
    let image = this.cache.get(path);
    if (image === undefined) {
      image = this.inner.load(path);
      this.cache.set(path, image);
    }
    return image;
  }
}

class AuditProxy implements ImageSource { // 记审计日志：谁读了哪张
  constructor(private readonly inner: ImageSource, private readonly user: User, private readonly log: string[]) {}

  load(path: string): Image {
    this.log.push(`${this.user.name} 读取 ${path}`);
    return this.inner.load(path);
  }
}

class AccessProxy implements ImageSource { // 保护代理：没权限就不往里转
  constructor(private readonly inner: ImageSource, private readonly user: User) {}

  load(path: string): Image {
    if (!this.user.canView(path)) throw new PermissionError(`${this.user.name} 无权查看 ${path}`);
    return this.inner.load(path);
  }
}

class LazyImage implements Image { // 虚代理：替一张图占位，第一次 show 才去拉
  private real: Image | undefined;

  constructor(private readonly path: string, private readonly source: ImageSource) {}

  show(): string {
    this.real ??= this.source.load(this.path);
    return this.real.show();
  }
}

// ---- 组装处：整个程序只有这里 new RemoteImageStore，页面和导出拿到的都是 ImageSource ----
const auditLog: string[] = [];
const remote = new RemoteImageStore();
const cache = new CachingProxy(remote); // 全站共享一份缓存，放在权限检查里面

function imageSourceFor(user: User): ImageSource { // 顺序：先查权限，再记日志，再查缓存，最后才拉
  return new AccessProxy(new AuditProxy(cache, user, auditLog), user);
}

// ---- 下面是页面和功能，只认 Image 和 ImageSource ----
function albumPage(paths: string[], source: ImageSource): Image[] { // 列表页：50 张卡片，一张都不拉
  return paths.map((p) => new LazyImage(p, source));
}

function click(card: Image): string { // 和 12.2 一样，点开就是 show
  return card.show();
}

function exportAlbum(paths: string[], source: ImageSource): Image[] { // 新同事的写法没变，只是手里的 source 变了
  return paths.map((p) => source.load(p));
}

const alice = new User("alice");
const bob = new User("bob");
const trip = Array.from({ length: 50 }, (_, i) => `public/trip/${String(i + 1).padStart(2, "0")}.jpg`);
const cards = albumPage(trip, imageSourceFor(bob));
console.log(`列表页：${cards.length} 张卡片，拉取 ${remote.fetches} 次`);
console.log(`bob 点开第 1 张：${click(cards[0])}，拉取 ${remote.fetches} 次`);
console.log(`bob 再点第 1 张：${click(cards[0])}，拉取 ${remote.fetches} 次`);
const aliceCards = albumPage(trip.slice(0, 1), imageSourceFor(alice));
console.log(`alice 点开同一张：${click(aliceCards[0])}，拉取 ${remote.fetches} 次`);

const secret = [1, 2, 3].map((i) => `private/alice/0${i}.jpg`);
try {
  exportAlbum(secret, imageSourceFor(bob));
} catch (error) {
  if (!(error instanceof PermissionError)) throw error;
  console.log(`bob 导出 alice 的私密相册：拒绝，${error.message}`);
}
console.log(`alice 导出自己的私密相册：${exportAlbum(secret, imageSourceFor(alice)).length} 张，拉取 ${remote.fetches} 次`);
console.log(`审计日志 ${auditLog.length} 条：`);
for (const line of auditLog) console.log(`  ${line}`);

// 输出：
// 列表页：50 张卡片，拉取 0 次
// bob 点开第 1 张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
// bob 再点第 1 张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
// alice 点开同一张：显示 public/trip/01.jpg（2048 KB），拉取 1 次
// bob 导出 alice 的私密相册：拒绝，bob 无权查看 private/alice/01.jpg
// alice 导出自己的私密相册：3 张，拉取 4 次
// 审计日志 5 条：
//   bob 读取 public/trip/01.jpg
//   alice 读取 public/trip/01.jpg
//   alice 读取 private/alice/01.jpg
//   alice 读取 private/alice/02.jpg
//   alice 读取 private/alice/03.jpg

看标红的四个类：三个代理和 RemoteImageStore 实现同一个 ImageSource，各自持有下一层；LazyImage 实现的是 Image，和真正的 Photo 站在一起。右上角的页面和导出功能只连着两个接口，左边的 image_source_for 是唯一连着 RemoteImageStore 的地方。

看顺序图：loop 里建 50 张卡片，一次都没拉；bob 点开第一张，调用依次经过权限、日志、缓存，到 RemoteImageStore 拉 1 次；alt 的另一支是 alice 之后点开同一张，经过她自己的权限和日志代理，到缓存就停了。bob 再点同一张卡片，LazyImage 手里已经有图，连代理都不经过。所以审计日志记的是对图片服务的每一次 load，不是每一次显示：bob 第二次点开不会再记一条。

输出里的拉取次数说明了两件事：列表页 50 张卡片拉取 0 次，是 LazyImage 的懒加载；三次点开同一张图只拉 1 次，靠的是 LazyImage 留住了图、CachingProxy 在两个用户之间共享了图。最后 alice 导出 3 张私密照片，拉取从 1 次变成 4 次。

逐条对照 12.3 的四处：

• 第 1 处，ImageStore 越塞越多：现在 RemoteImageStore 只有拉取，缓存、日志各在自己的类里。load 的签名回到了只有 path，用户由代理在创建时拿着，页面调 load 不用传。
• 第 2 处，懒加载改页面：album_page 只是把 Photo 换成了 LazyImage，返回的还是 Image 列表；click 和 12.2 一字不差，还是 card.show()。
• 第 3 处，每个调用处判权限：页面里一句 can_view 都没有了，权限只在 AccessProxy 里判一次。
• 第 4 处，导出事故：新同事的 export_album 和 12.3 一样只管 load，可他手里的 source 是 image_source_for(bob) 给的，最外层是 AccessProxy，输出里 bob 的导出被拒了。

要说清楚的是，导出功能被拦住，不是因为代理让人写不错，而是因为组装处不再把裸的 RemoteImageStore 交出去。这条规则住在 image_source_for 和 _remote、_cache 这两个模块级变量上，两个一样没有保护：Python 的下划线只是约定，假设这段代码放在 album.py 里，谁硬要 from album import _remote 或 _cache 照样拿得到，拿到 _cache 一样绕过了权限；TS 里（文件是模块时）只要不 export 这两个变量，别的文件就引用不到。真实项目要把它写下来，并让它可检查：图片服务单独一个模块，只公开 image_source_for；再加一条代码评审规则或测试，只允许组装处创建 RemoteImageStore。

代理的顺序也有讲究，规则同样写在组装处的注释里。缓存是全站共享的，必须放在权限检查里面：要是把 CachingProxy 放到最外层，alice 看过的私密照片留在缓存里，bob 再要同一个路径时，缓存直接给出去，权限检查根本轮不到。审计日志放在权限检查里面，记下的是读成功的；想连被拒的尝试也记下来，就把 AuditProxy 挪到 AccessProxy 外面。

12.5 模式定义

GoF 对代理的定义是：为其他对象提供一种代理以控制对这个对象的访问。

代理和它代替的真实对象（real subject）实现同一个接口，调用方分不出手里拿的是哪个。代理在把调用转给真实对象之前，可以决定转不转（没权限就不转）、什么时候转（第一次真正用到才转）、转之前先做什么（记日志、查缓存）。GoF 按目的列了几种常见的代理：

• 虚代理（virtual proxy）：真实对象创建起来很贵，先放一个替身，真用到时再创建。本章的 LazyImage。
• 保护代理（protection proxy）：先查调用方有没有权限，再决定转不转。本章的 AccessProxy。
• 远程代理（remote proxy）：真实对象在另一个进程或另一台机器上，代理在本地冒充它，把方法调用变成网络请求。本章没写，RPC 框架生成的客户端就是这种。
• 智能引用（smart reference）：访问真实对象时顺带做点额外的事，GoF 举的是引用计数、首次引用时把持久对象装进内存。本章的 AuditProxy、CachingProxy 算这一类，缓存常被单独叫作缓存代理。

GoF 的图里代理直接持有 RealSubject；本章的代理持有的是 ImageSource 接口，所以几层代理可以串起来。

角色	在本章案例里是谁	职责
主题（Subject）	ImageSource；虚代理那边是 Image	真实对象和代理共同的接口，调用方只认它
真实主题（RealSubject）	RemoteImageStore；虚代理那边是 Photo	真正干活：拉图 / 显示图
代理（Proxy）	AccessProxy、AuditProxy、CachingProxy、LazyImage	实现同一个接口，持有下一层，决定转不转、何时转、转前做什么
客户（Client）	album_page、click、export_album	只用接口，不知道手里是不是代理、有几层

常见场景：

• 延迟加载。ORM 里的关联对象，比如 order.customer，往往在第一次访问时才查库；12.6 例一也是这一类。
• 权限控制。给内部服务包一层，按调用方身份决定放不放行，被保护的服务本身不用知道有权限这回事。
• 远程调用桩（stub）。gRPC 这类 RPC 框架按接口定义生成客户端，调用写起来像本地方法，背后是序列化后发网络请求。
• 缓存。给慢的查询包一层，命中就不往下走。
• 引用计数。记下有几处在用真实对象，没人用了就释放，C++ 的 std::shared_ptr 就是这个思路。

12.6 真实系统里的例子

例一：Django 的 SimpleLazyObject（Python）

Django 里每个视图都能用 request.user 拿到当前登录用户，但不是每个视图都用得着它。AuthenticationMiddleware 在请求进来时做的大致是一句 request.user = SimpleLazyObject(lambda: get_user(request))：先放一个替身，等视图第一次读 request.user 的某个属性，替身才调 get_user，按 session 里的用户 id 去查库（较新的版本另外提供了异步用的 request.auser）。不碰用户的视图，一次查询都不会发生。SimpleLazyObject 在 django.utils.functional 里，简化版：

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any

_empty = object()  # Django 里也用一个哨兵对象表示"还没求值"


class SimpleLazyObject:  # 大致相当于 django.utils.functional.SimpleLazyObject，只保留"读属性"这一条路
    def __init__(self, func: Callable[[], Any]) -> None:
        self.__dict__["_setupfunc"] = func  # 绕开 __setattr__，真实实现也有类似处理
        self.__dict__["_wrapped"] = _empty

    def __getattr__(self, name: str) -> Any:  # 在实例上找不到的属性才会走到这里
        if self._wrapped is _empty:
            self.__dict__["_wrapped"] = self._setupfunc()  # 第一次：真正创建对象
        return getattr(self._wrapped, name)  # 以后：原样转发


@dataclass
class User:  # 大致相当于 django.contrib.auth.models.User
    username: str
    is_authenticated: bool = True


@dataclass
class Request:
    session: dict[str, int]
    user: Any = None


queries = 0


def get_user(request: Request) -> User:  # 大致相当于 django.contrib.auth.get_user：按 session 里的 id 查库
    global queries
    queries += 1
    print(f"  查询了一次数据库：id = {request.session['_auth_user_id']}")
    return User("alice")


def authentication_middleware(request: Request) -> None:  # AuthenticationMiddleware 做的就是这一句
    request.user = SimpleLazyObject(lambda: get_user(request))


def home(request: Request) -> str:  # 不碰 request.user 的视图
    return "首页"


def profile(request: Request) -> str:  # 读了两次 request.user 的属性
    return f"{request.user.username}，已登录：{request.user.is_authenticated}"


if __name__ == "__main__":
    for view in (home, profile):
        request = Request(session={"_auth_user_id": 42})
        authentication_middleware(request)
        print(f"{view.__name__}：{view(request)}，累计查询 {queries} 次")

# 输出：
# home：首页，累计查询 0 次
#   查询了一次数据库：id = 42
# profile：alice，已登录：True，累计查询 1 次

SimpleLazyObject 对应本章的 LazyImage：工厂函数相当于 source.load(path)，_wrapped 相当于 real。区别在于 Python 的 __getattr__ 能把任何属性读取都转过去，用不着先定义一个 Image 接口、逐个方法写转发。真实的版本做得更全：__str__、__eq__ 这些特殊方法也一个个转发，连 __class__ 都转，所以 isinstance(request.user, User) 能成立。

例二：JavaScript 的 Proxy 和 Vue 3 的 reactive（TypeScript）

JavaScript 从 ES2015 起内置了 Proxy：new Proxy(target, handler) 返回一个替身对象，对它的读属性、写属性、删属性等操作，都会先进 handler 里对应的函数（叫 trap），由你决定怎么转给 target。Vue 3 的 reactive(obj) 返回的就是这样一个 Proxy。真实的实现在 get 里记下哪段渲染或计算读了这个属性，在 set 里只通知读过它的那些，还要排队批量执行，比这里复杂得多。下面的模型只演示拦截：set 时值变了就打印“触发更新”并重新渲染。

// 模拟 Vue 3 的 reactive()：真实实现在 @vue/reactivity 里，get 时记下"谁读了这个属性"（track），
// set 时只通知读过它的副作用、并且排队批量执行（trigger + 调度器）。这里只演示 Proxy 的拦截本身。
function reactive<T extends object>(target: T, onChange: () => void): T {
  return new Proxy(target, {
    get(obj, key, receiver) {
      return Reflect.get(obj, key, receiver); // Vue 在这里 track
    },
    set(obj, key, value, receiver) {
      const old: unknown = Reflect.get(obj, key, receiver);
      const ok = Reflect.set(obj, key, value, receiver);
      if (!Object.is(old, value)) { // 值没变就不触发，Vue 也是这样判断的
        console.log(`触发更新：${String(key)} ${String(old)} → ${String(value)}`);
        onChange(); // Vue 在这里 trigger
      }
      return ok;
    },
  });
}

const raw = { title: "厦门旅行", likes: 0 };
const state = reactive(raw, render);

function render(): void { // 相当于组件的渲染函数
  console.log(`  渲染：${state.title}，${state.likes} 个赞`);
}

render();
state.likes += 1; // 经过代理：先 get 再 set，set 里触发更新
state.title = "厦门旅行"; // 值没变，不触发
raw.likes = 99; // 改原对象：不经过代理，页面不知道
console.log(`原对象已是 ${raw.likes}，页面还停在上一次渲染`);
state.title = "厦门 · 鼓浪屿";

// 输出：
//   渲染：厦门旅行，0 个赞
// 触发更新：likes 0 → 1
//   渲染：厦门旅行，1 个赞
// 原对象已是 99，页面还停在上一次渲染
// 触发更新：title 厦门旅行 → 厦门 · 鼓浪屿
//   渲染：厦门 · 鼓浪屿，99 个赞

reactive 返回的 Proxy 在结构上和本章的代理一样，raw 就是真实对象；按意图却更接近装饰器：它给对象加上了“改了就重新渲染”这项功能，去掉它页面就不会自动更新，按 12.8 的判别句算功能少了一样。JS 的 Proxy 只是拦截机制，写出来的可以是代理，也可以是装饰器。倒数第二段值得注意：直接改 raw 不经过代理，页面不知道；Vue 的文档里也提醒过，只有代理是响应式的，改原对象不会触发更新。这和 12.3 的导出事故是一回事：谁拿到了真实对象，谁就绕过了代理，所以要管住的是真实对象交给谁。

12.7 什么时候别用

• 只是想加功能，不是控制访问。比如给导出的图片打水印，去掉它图片就少了水印，这是在加功能，用第 10 章的装饰器，名字也该叫 Watermarked。叫成 WatermarkProxy，读代码的人会以为它在把关，去里面找权限逻辑。
• 替身藏得太好，出了事找不到。代理常常只有一层，不像第 10 章那样套七层，可调用方本来就不知道它在，问题也就出在这里。LazyImage 把拉图从列表页挪到了点开那一刻，对象存储挂了，异常就从 click 里冒出来，离创建卡片的代码很远；后来有人加了“全部下载”按钮，循环调 50 次 show，就是 50 次一张一张的拉取，懒加载把一次批量拉取的机会藏了起来，ORM 里的 N+1 查询也是这么来的。用了虚代理，要在名字或注释里写明“第一次访问会拉图”，批量场景另给一个批量接口。
• 热路径上每次都过一遍代理。夜里生成缩略图的任务要把 10 万张图各 load 一次，如果它也从 image_source_for 拿图片服务，就是 10 万次权限判断、10 万条审计日志，而 CachingProxy 不会淘汰，10 万张原图全留在内存里；真实的缓存要有容量上限和淘汰策略，比如只留最近用过的若干张。这个任务是系统自己跑的，用不着这些把关。给它在组装处单独登记一个只有 RemoteImageStore 的入口，取个一看就知道的名字，比如 system_image_source()，裸的真实对象只从这一处交出去。

12.8 容易混淆的模式

模式	意图	区别
代理（Proxy）	用一个替身控制对原对象的访问	同接口、持有原对象；不管谁拿到都套着，调用方不知道有它、也选不掉；去掉它，功能一样，只是访问没人把关了
装饰器（Decorator）	不改原对象，给它叠加功能	结构几乎相同；按需要挑哪几层、怎么叠（第 10 章按租户在 TENANT_STORAGE 里配）；去掉它，功能少了一样，见第 10 章
适配器（Adapter）	让接口对不上的类按调用方要的接口工作	适配器换接口，代理不换接口，见第 9 章
外观（Facade）	给一组子系统一个简单入口	外观定义新的、更简单的接口，背后是好几个对象；代理和原对象接口相同，背后是一个对象，见第 11 章

和第 10 章一样，判别装饰器和代理最实用的一句：“去掉它，功能少了”是装饰器；“去掉它，功能一样，只是访问没人把关了”是代理。拿本章的类试一试：去掉 AccessProxy，图片照样显示，只是 bob 也能看 alice 的私密照片了；去掉 LazyImage，页面照样显示，只是 50 张图什么时候拉没人管了；去掉 AuditProxy，图片照样显示，只是谁读了哪张没人记了。这三个是代理。

CachingProxy 落在两者中间：它的代码和第 10 章的 Cached 几乎一行不差，第 10 章的表把“不缓存了”算作功能少了一样。叫它什么，看谁组装、为了什么：第 10 章的缓存是租户 C 开的一项功能，按租户叠；这里的缓存是图片服务统一挂上的，页面不知道也选不了，目的是少拉几次对象存储。结构一样时，名字是在告诉读代码的人：这一层是来把关的，还是来加功能的。

12.9 练一练

系统里发邮件的是 EmailGateway，只有一个方法 send(to, body)。已有一个真实实现 SmtpGateway，send 时打印一行“SMTP 发出”。邮件服务商限制每分钟最多发 N 封，超了会封号。

请用代理实现：

• RateLimitProxy：同样实现 EmailGateway，持有下一层。最近 60 秒内已经发了 N 封时，抛一个 RateLimitExceeded，不往下转。时间不要直接读系统时钟，构造时传入一个返回当前秒数的函数，方便测试。
• LoggingProxy：每次 send 记一条日志，成功记“成功 地址”，被限流记“被限流 地址”，然后把异常继续抛出去。
• 把两层组合起来，N 取 2：在第 0、10、20、65 秒各发一封，打印每次的结果和最后的日志。Python 和 TS 各写一版。

提示：两层谁在外面，日志里才会有“被限流”那一条？先预测，再把顺序对调跑一次验证。

12.10 小结

• 代理和真实对象实现同一个接口，调用方分不出来；代理决定转不转、什么时候转、转之前先做什么。
• 常见的几种：虚代理推迟创建，保护代理先查权限，远程代理把调用变成网络请求，智能引用（包括缓存）在访问时顺带做事。几种代理可以串起来，顺序写在组装处：共享缓存必须放在权限检查里面。
• 代理守得住，靠的是真实对象只从组装处交出去。谁拿到了裸的真实对象，谁就绕过了所有代理，这条规则要写下来，并且能检查。
• 虚代理把开销挪到第一次访问，也会把批量拉取的机会藏起来；热路径和批处理任务不要无差别地套代理。
• 和装饰器结构几乎一样：去掉后功能少了的是装饰器，去掉后功能一样、只是访问没人把关了的是代理。