第 4 章 · 抽象工厂（Abstract Factory）

一句话：用一个工厂对象成套地创建一组配合使用的对象，换一个工厂就换掉整套。 解决的痛：几个必须配套的对象在不同地方各自按类型 if 出来，加一种类型要改每一处，配错一个也没有任何报错。

4.1 一个场景

还是那家做在线课程的公司。用户买课要付钱，目前支持两个支付渠道：支付宝和微信支付。

每个渠道都要用到三个对象，它们由不同的同事按渠道封装好了：

PaymentClient：向渠道发起一笔支付，方法是 pay(order_id, amount)。
RefundClient：向渠道申请退款，方法是 refund(order_id, amount)。
WebhookVerifier：校验回调（webhook）的签名。用户付完钱，渠道会回调我们的接口告诉我们“这笔钱到了”，我们必须先验签名，确认这个请求真是渠道发来的，不是有人伪造的，方法是 verify(payload, signature)。

这三个对象是配套的：支付宝的支付客户端发起的订单，只有支付宝的退款客户端能退，回调也只有支付宝的验签器能验。签名算法和密钥各家都不一样，比如支付宝用 RSA2 签名，微信支付（APIv2）用商户密钥算 HMAC-SHA256。真实项目里它们背后是各家的 SDK，本章用 print 和字符串比较代替。

你要写三个业务函数：下单时调用的 checkout，用户申请退款时调用的 refund_order，渠道回调进来时调用的 handle_webhook。在真实项目里，它们分别在 checkout.py、refunds.py、webhooks.py 三个文件里，由不同的人维护；本章为了能直接运行，把它们放在一个文件里。

两个渠道各一套三件套

两个渠道各有一套三件套，横着看是同一种产品，竖着看必须来自同一个渠道。

4.2 先按直觉写

和第 3 章一样，最直接的写法是每个函数按订单的渠道判断，该用哪个渠道的对象就 new 哪个：

class AlipayPaymentClient:
    def pay(self, order_id: str, amount: int) -> None:
        print(f"[支付宝] 订单 {order_id} 发起支付 {amount} 元")


class AlipayRefundClient:
    def refund(self, order_id: str, amount: int) -> None:
        print(f"[支付宝] 订单 {order_id} 退款 {amount} 元")


class AlipayWebhookVerifier:
    def verify(self, payload: str, signature: str) -> bool:
        return signature == f"RSA2:{payload}"  # 模拟：真实做法是用支付宝公钥验 RSA2 签名


class WechatPaymentClient:
    def pay(self, order_id: str, amount: int) -> None:
        print(f"[微信支付] 订单 {order_id} 发起支付 {amount} 元")


class WechatRefundClient:
    def refund(self, order_id: str, amount: int) -> None:
        print(f"[微信支付] 订单 {order_id} 退款 {amount} 元")


class WechatWebhookVerifier:
    def verify(self, payload: str, signature: str) -> bool:
        return signature == f"HMAC-SHA256:{payload}"  # 模拟：真实做法是用商户密钥重算签名再比对


# 调用点一：下单支付（真实项目里在 checkout.py）
def checkout(channel: str, order_id: str, amount: int) -> None:
    if channel == "alipay":
        AlipayPaymentClient().pay(order_id, amount)
    elif channel == "wechat":
        WechatPaymentClient().pay(order_id, amount)
    else:
        raise ValueError(f"未知渠道：{channel}")


# 调用点二：申请退款（真实项目里在 refunds.py）
def refund_order(channel: str, order_id: str, amount: int) -> None:
    if channel == "alipay":
        AlipayRefundClient().refund(order_id, amount)
    elif channel == "wechat":
        WechatRefundClient().refund(order_id, amount)
    else:
        raise ValueError(f"未知渠道：{channel}")


# 调用点三：处理支付回调（真实项目里在 webhooks.py）
def handle_webhook(channel: str, payload: str, signature: str) -> None:
    if channel == "alipay":
        ok = AlipayWebhookVerifier().verify(payload, signature)
    elif channel == "wechat":
        ok = WechatWebhookVerifier().verify(payload, signature)
    else:
        raise ValueError(f"未知渠道：{channel}")
    if ok:
        print(f"[回调] 订单 {payload} 签名有效，标记为已支付")
    else:
        print(f"[回调] 订单 {payload} 签名无效，按伪造请求拒绝")


if __name__ == "__main__":
    checkout("alipay", "1001", 199)
    handle_webhook("alipay", "1001", "RSA2:1001")
    checkout("wechat", "1002", 99)
    handle_webhook("wechat", "1002", "HMAC-SHA256:1002")
    refund_order("wechat", "1002", 99)

# 输出：
# [支付宝] 订单 1001 发起支付 199 元
# [回调] 订单 1001 签名有效，标记为已支付
# [微信支付] 订单 1002 发起支付 99 元
# [回调] 订单 1002 签名有效，标记为已支付
# [微信支付] 订单 1002 退款 99 元

TypeScript 版结构一样：

class AlipayPaymentClient {
  pay(orderId: string, amount: number): void {
    console.log(`[支付宝] 订单 ${orderId} 发起支付 ${amount} 元`);
  }
}

class AlipayRefundClient {
  refund(orderId: string, amount: number): void {
    console.log(`[支付宝] 订单 ${orderId} 退款 ${amount} 元`);
  }
}

class AlipayWebhookVerifier {
  verify(payload: string, signature: string): boolean {
    return signature === `RSA2:${payload}`; // 模拟：真实做法是用支付宝公钥验 RSA2 签名
  }
}

class WechatPaymentClient {
  pay(orderId: string, amount: number): void {
    console.log(`[微信支付] 订单 ${orderId} 发起支付 ${amount} 元`);
  }
}

class WechatRefundClient {
  refund(orderId: string, amount: number): void {
    console.log(`[微信支付] 订单 ${orderId} 退款 ${amount} 元`);
  }
}

class WechatWebhookVerifier {
  verify(payload: string, signature: string): boolean {
    return signature === `HMAC-SHA256:${payload}`; // 模拟：真实做法是用商户密钥重算签名再比对
  }
}

// 调用点一：下单支付（真实项目里在 checkout.ts）
function checkout(channel: string, orderId: string, amount: number): void {
  if (channel === "alipay") {
    new AlipayPaymentClient().pay(orderId, amount);
  } else if (channel === "wechat") {
    new WechatPaymentClient().pay(orderId, amount);
  } else {
    throw new Error(`未知渠道：${channel}`);
  }
}

// 调用点二：申请退款（真实项目里在 refunds.ts）
function refundOrder(channel: string, orderId: string, amount: number): void {
  if (channel === "alipay") {
    new AlipayRefundClient().refund(orderId, amount);
  } else if (channel === "wechat") {
    new WechatRefundClient().refund(orderId, amount);
  } else {
    throw new Error(`未知渠道：${channel}`);
  }
}

// 调用点三：处理支付回调（真实项目里在 webhooks.ts）
function handleWebhook(channel: string, payload: string, signature: string): void {
  let ok: boolean;
  if (channel === "alipay") {
    ok = new AlipayWebhookVerifier().verify(payload, signature);
  } else if (channel === "wechat") {
    ok = new WechatWebhookVerifier().verify(payload, signature);
  } else {
    throw new Error(`未知渠道：${channel}`);
  }
  if (ok) {
    console.log(`[回调] 订单 ${payload} 签名有效，标记为已支付`);
  } else {
    console.log(`[回调] 订单 ${payload} 签名无效，按伪造请求拒绝`);
  }
}

checkout("alipay", "1001", 199);
handleWebhook("alipay", "1001", "RSA2:1001");
checkout("wechat", "1002", 99);
handleWebhook("wechat", "1002", "HMAC-SHA256:1002");
refundOrder("wechat", "1002", 99);

// 输出：
// [支付宝] 订单 1001 发起支付 199 元
// [回调] 订单 1001 签名有效，标记为已支付
// [微信支付] 订单 1002 发起支付 99 元
// [回调] 订单 1002 签名有效，标记为已支付
// [微信支付] 订单 1002 退款 99 元

三个函数各管一件事，每个都能从上往下读懂。两个渠道、三个调用点，每个 if 只有两个分支，一眼就能核对。大多数团队接第二个支付渠道时都是这么写的，它能稳定跑很久。

4.3 需求变了

公司开始卖海外课程，产品经理要求：接入 Stripe，海外用户用信用卡付款，退款和回调验签也走 Stripe。

要改的地方有四处：

新增 StripePaymentClient、StripeRefundClient、StripeWebhookVerifier 三个类。这一处任何写法都躲不掉。
checkout.py 的 checkout：加一个 elif channel == "stripe" 分支，创建 StripePaymentClient。
refunds.py 的 refund_order：加同样的分支，创建 StripeRefundClient。
webhooks.py 的 handle_webhook：加同样的分支，创建 StripeWebhookVerifier。

TS 版是同样的四处。第 2 到 4 处在三个文件里，最顺手的做法是复制上面那个分支、改一下条件。问题就出在这里。下面是一次真实可能发生的提交，三个分支各是从哪里复制来的，就保留了哪里的类名：

# checkout.py：改对了
elif channel == "stripe":
    StripePaymentClient().pay(order_id, amount)

# refunds.py：复制了支付宝那个分支，只改了条件
elif channel == "stripe":
    AlipayRefundClient().refund(order_id, amount)

# webhooks.py：复制了微信那个分支，只改了条件
elif channel == "stripe":
    ok = WechatWebhookVerifier().verify(payload, signature)

这段代码能通过类型检查，也能正常启动：AlipayRefundClient 本来就有 refund 方法，WechatWebhookVerifier 本来就有 verify 方法，签名都对得上。上线后一笔 Stripe 订单走一遍，日志是这样的：

[Stripe] 订单 1003 发起支付 30 元
[回调] 订单 1003 签名无效，按伪造请求拒绝
[支付宝] 订单 1003 退款 30 元

第一行没问题，钱扣了。第二行，Stripe 发来的回调带的是 Stripe 的签名，微信的验签器按微信的规则去比对，当然对不上，于是每一个 Stripe 回调都被当成伪造请求拒掉了。用户付了钱，订单一直停在“待支付”，课程打不开；Stripe 收到拒绝会隔一段时间重发，重发多少次都一样。第三行，用户来退款，请求发到了支付宝，支付宝从没见过这个订单号，返回“交易不存在”，钱退不出去。

这种错很难在上线前发现。支付宝和微信的测试全部通过；做 Stripe 的同事在沙箱（sandbox）里测了下单，扣款成功就提交了。回调要等渠道真的打过来，退款要等用户真的来退，问题要等客服接到投诉才暴露。三个文件各自选对象，没有任何代码保证这三处选的是同一个渠道，只能靠人逐行核对。

还有一种漏法：refund_order 那个分支忘了加。这次会报错，但不是在启动时，而是在第一个 Stripe 用户申请退款时抛出“未知渠道：stripe”。

问题的形状是：每加一个渠道要改三处 if，而一个渠道配套的三个对象分散在三处各自创建，配错了编译器不报错，程序也照常跑。

朴素写法：四处要改的位置

图中 ①–④ 就是上面列出的四处，三个业务函数各自连着所有渠道的类。

4.4 用抽象工厂模式重写

既然三个对象必须来自同一个渠道，就让同一个对象负责把它们三个造出来。定义一个 PaymentGateway（支付网关）接口，它有三个创建方法：create_payment()、create_refund()、create_verifier()。每个渠道实现一个 gateway，产出自己那一整套。业务函数不再收渠道名字符串，而是收一个 gateway，需要哪个对象就向它要：

from abc import ABC, abstractmethod
from typing import Protocol


class PaymentClient(Protocol):  # 三种产品的接口
    def pay(self, order_id: str, amount: int) -> None: ...


class RefundClient(Protocol):
    def refund(self, order_id: str, amount: int) -> None: ...


class WebhookVerifier(Protocol):
    def verify(self, payload: str, signature: str) -> bool: ...


class AlipayPaymentClient:
    def pay(self, order_id: str, amount: int) -> None:
        print(f"[支付宝] 订单 {order_id} 发起支付 {amount} 元")


class AlipayRefundClient:
    def refund(self, order_id: str, amount: int) -> None:
        print(f"[支付宝] 订单 {order_id} 退款 {amount} 元")


class AlipayWebhookVerifier:
    def verify(self, payload: str, signature: str) -> bool:
        return signature == f"RSA2:{payload}"


class WechatPaymentClient:
    def pay(self, order_id: str, amount: int) -> None:
        print(f"[微信支付] 订单 {order_id} 发起支付 {amount} 元")


class WechatRefundClient:
    def refund(self, order_id: str, amount: int) -> None:
        print(f"[微信支付] 订单 {order_id} 退款 {amount} 元")


class WechatWebhookVerifier:
    def verify(self, payload: str, signature: str) -> bool:
        return signature == f"HMAC-SHA256:{payload}"


class StripePaymentClient:
    def pay(self, order_id: str, amount: int) -> None:
        print(f"[Stripe] 订单 {order_id} 发起支付 {amount} 元")


class StripeRefundClient:
    def refund(self, order_id: str, amount: int) -> None:
        print(f"[Stripe] 订单 {order_id} 退款 {amount} 元")


class StripeWebhookVerifier:
    def verify(self, payload: str, signature: str) -> bool:
        return signature == f"v1={payload}"  # 模拟：Stripe 用签名密钥算 HMAC，放在 v1= 后面


class PaymentGateway(ABC):  # 抽象工厂：一个渠道的三件套都从这里拿
    @abstractmethod
    def create_payment(self) -> PaymentClient: ...

    @abstractmethod
    def create_refund(self) -> RefundClient: ...

    @abstractmethod
    def create_verifier(self) -> WebhookVerifier: ...


class AlipayGateway(PaymentGateway):
    def create_payment(self) -> PaymentClient:
        return AlipayPaymentClient()

    def create_refund(self) -> RefundClient:
        return AlipayRefundClient()

    def create_verifier(self) -> WebhookVerifier:
        return AlipayWebhookVerifier()


class WechatGateway(PaymentGateway):
    def create_payment(self) -> PaymentClient:
        return WechatPaymentClient()

    def create_refund(self) -> RefundClient:
        return WechatRefundClient()

    def create_verifier(self) -> WebhookVerifier:
        return WechatWebhookVerifier()


class StripeGateway(PaymentGateway):  # 新渠道：Stripe 的配套关系只写在这里
    def create_payment(self) -> PaymentClient:
        return StripePaymentClient()

    def create_refund(self) -> RefundClient:
        return StripeRefundClient()

    def create_verifier(self) -> WebhookVerifier:
        return StripeWebhookVerifier()


# 调用点一：下单支付
def checkout(gateway: PaymentGateway, order_id: str, amount: int) -> None:
    gateway.create_payment().pay(order_id, amount)


# 调用点二：申请退款
def refund_order(gateway: PaymentGateway, order_id: str, amount: int) -> None:
    gateway.create_refund().refund(order_id, amount)


# 调用点三：处理支付回调
def handle_webhook(gateway: PaymentGateway, payload: str, signature: str) -> None:
    if gateway.create_verifier().verify(payload, signature):
        print(f"[回调] 订单 {payload} 签名有效，标记为已支付")
    else:
        print(f"[回调] 订单 {payload} 签名无效，按伪造请求拒绝")


# 程序入口：把订单上的渠道名换成 gateway，全项目只有这一处
GATEWAYS: dict[str, type[PaymentGateway]] = {
    "alipay": AlipayGateway,
    "wechat": WechatGateway,
    "stripe": StripeGateway,
}

if __name__ == "__main__":
    orders = [("alipay", "1001", "RSA2:1001"), ("wechat", "1002", "HMAC-SHA256:1002"),
              ("stripe", "1003", "v1=1003")]
    for channel, order_id, signature in orders:
        gateway = GATEWAYS[channel]()
        checkout(gateway, order_id, 30)
        handle_webhook(gateway, order_id, signature)
    refund_order(GATEWAYS["stripe"](), "1003", 30)

# 输出：
# [支付宝] 订单 1001 发起支付 30 元
# [回调] 订单 1001 签名有效，标记为已支付
# [微信支付] 订单 1002 发起支付 30 元
# [回调] 订单 1002 签名有效，标记为已支付
# [Stripe] 订单 1003 发起支付 30 元
# [回调] 订单 1003 签名有效，标记为已支付
# [Stripe] 订单 1003 退款 30 元

TS 版的产品和工厂都用 interface，映射表的值类型写成构造签名 new () => PaymentGateway：

interface PaymentClient { // 三种产品的接口
  pay(orderId: string, amount: number): void;
}

interface RefundClient {
  refund(orderId: string, amount: number): void;
}

interface WebhookVerifier {
  verify(payload: string, signature: string): boolean;
}

class AlipayPaymentClient implements PaymentClient {
  pay(orderId: string, amount: number): void {
    console.log(`[支付宝] 订单 ${orderId} 发起支付 ${amount} 元`);
  }
}

class AlipayRefundClient implements RefundClient {
  refund(orderId: string, amount: number): void {
    console.log(`[支付宝] 订单 ${orderId} 退款 ${amount} 元`);
  }
}

class AlipayWebhookVerifier implements WebhookVerifier {
  verify(payload: string, signature: string): boolean {
    return signature === `RSA2:${payload}`;
  }
}

class WechatPaymentClient implements PaymentClient {
  pay(orderId: string, amount: number): void {
    console.log(`[微信支付] 订单 ${orderId} 发起支付 ${amount} 元`);
  }
}

class WechatRefundClient implements RefundClient {
  refund(orderId: string, amount: number): void {
    console.log(`[微信支付] 订单 ${orderId} 退款 ${amount} 元`);
  }
}

class WechatWebhookVerifier implements WebhookVerifier {
  verify(payload: string, signature: string): boolean {
    return signature === `HMAC-SHA256:${payload}`;
  }
}

class StripePaymentClient implements PaymentClient {
  pay(orderId: string, amount: number): void {
    console.log(`[Stripe] 订单 ${orderId} 发起支付 ${amount} 元`);
  }
}

class StripeRefundClient implements RefundClient {
  refund(orderId: string, amount: number): void {
    console.log(`[Stripe] 订单 ${orderId} 退款 ${amount} 元`);
  }
}

class StripeWebhookVerifier implements WebhookVerifier {
  verify(payload: string, signature: string): boolean {
    return signature === `v1=${payload}`; // 模拟：Stripe 用签名密钥算 HMAC，放在 v1= 后面
  }
}

interface PaymentGateway { // 抽象工厂：一个渠道的三件套都从这里拿
  createPayment(): PaymentClient;
  createRefund(): RefundClient;
  createVerifier(): WebhookVerifier;
}

class AlipayGateway implements PaymentGateway {
  createPayment(): PaymentClient {
    return new AlipayPaymentClient();
  }

  createRefund(): RefundClient {
    return new AlipayRefundClient();
  }

  createVerifier(): WebhookVerifier {
    return new AlipayWebhookVerifier();
  }
}

class WechatGateway implements PaymentGateway {
  createPayment(): PaymentClient {
    return new WechatPaymentClient();
  }

  createRefund(): RefundClient {
    return new WechatRefundClient();
  }

  createVerifier(): WebhookVerifier {
    return new WechatWebhookVerifier();
  }
}

class StripeGateway implements PaymentGateway { // 新渠道：Stripe 的配套关系只写在这里
  createPayment(): PaymentClient {
    return new StripePaymentClient();
  }

  createRefund(): RefundClient {
    return new StripeRefundClient();
  }

  createVerifier(): WebhookVerifier {
    return new StripeWebhookVerifier();
  }
}

// 调用点一：下单支付
function checkout(gateway: PaymentGateway, orderId: string, amount: number): void {
  gateway.createPayment().pay(orderId, amount);
}

// 调用点二：申请退款
function refundOrder(gateway: PaymentGateway, orderId: string, amount: number): void {
  gateway.createRefund().refund(orderId, amount);
}

// 调用点三：处理支付回调
function handleWebhook(gateway: PaymentGateway, payload: string, signature: string): void {
  if (gateway.createVerifier().verify(payload, signature)) {
    console.log(`[回调] 订单 ${payload} 签名有效，标记为已支付`);
  } else {
    console.log(`[回调] 订单 ${payload} 签名无效，按伪造请求拒绝`);
  }
}

// 程序入口：把订单上的渠道名换成 gateway，全项目只有这一处
const GATEWAYS: Record<string, new () => PaymentGateway> = {
  alipay: AlipayGateway,
  wechat: WechatGateway,
  stripe: StripeGateway,
};

const orders: [string, string, string][] = [
  ["alipay", "1001", "RSA2:1001"], ["wechat", "1002", "HMAC-SHA256:1002"],
  ["stripe", "1003", "v1=1003"],
];
for (const [channel, orderId, signature] of orders) {
  const gateway = new GATEWAYS[channel]();
  checkout(gateway, orderId, 30);
  handleWebhook(gateway, orderId, signature);
}
refundOrder(new GATEWAYS["stripe"](), "1003", 30);

// 输出：
// [支付宝] 订单 1001 发起支付 30 元
// [回调] 订单 1001 签名有效，标记为已支付
// [微信支付] 订单 1002 发起支付 30 元
// [回调] 订单 1002 签名有效，标记为已支付
// [Stripe] 订单 1003 发起支付 30 元
// [回调] 订单 1003 签名有效，标记为已支付
// [Stripe] 订单 1003 退款 30 元

抽象工厂版：PaymentGateway 与三个产品接口

看标红的 StripeGateway：新渠道就是这一个类，业务函数只认 PaymentGateway。

逐条对照 4.3 的四处改动：

第 1 处，Stripe 的三个类：还是要写，这是新渠道本身的代码。
第 2 到 4 处，三个业务函数里的 Stripe 分支：不用加了。三个函数里没有任何按渠道的判断，加 Stripe 时一行没动。取而代之的是新增 StripeGateway 一个类，再在入口的 GATEWAYS 里加一行。这一行和简单工厂里加一个分支是一回事，任何写法都要在某个地方登记新渠道，抽象工厂省不掉它。
配错的问题：业务函数手里只有一个 gateway，refund_order 拿到的退款客户端和 handle_webhook 拿到的验签器，都出自 checkout 用的同一个 gateway。想把 Stripe 的支付和微信的验签配在一起，业务代码里已经没有地方能写出这种组合。“Stripe 用哪三个类”只写在 StripeGateway 里，三行 return 上下挨着，代码评审时一眼就能看出有没有写错，不用在三个文件之间来回核对。真实项目里，商户号、密钥这类一个渠道共用的配置也放进 gateway，三个对象从同一个地方拿，又少了一种配错的可能。
漏改的问题：StripeGateway 忘了写 create_refund，Python 在第一次执行 StripeGateway() 时就抛出 TypeError，提示有抽象方法没有实现，也就是第一笔 Stripe 下单时，在沙箱里测一次下单就能发现；TS 更早，编译时就报 StripeGateway 没有正确实现 PaymentGateway。原来要等第一个用户来退款才暴露的错，现在提前到了下单或编译这一步。

抽象工厂并不能防住所有错误：要是有人在 StripeGateway.create_verifier 里写了 return WechatWebhookVerifier()，它照样能跑。区别在于这个错只可能出在一个类的三行里，而不是散在三个文件中。代价也很明显：多了一个工厂接口和三个 gateway 类，读代码的人要多跳一层才知道 create_refund() 返回的是谁。

4.5 模式定义

GoF 对抽象工厂的定义是：提供一个创建一系列相关或相互依赖对象的接口，而无需指定它们具体的类。

换成本章的话说：“一系列相关的对象”就是一个渠道的三件套，这样一组必须配套使用的对象通常叫一个产品族（product family）。PaymentGateway 就是那个“接口”，它只说“我能给你一个支付客户端、一个退款客户端、一个验签器”，不说是哪个渠道的。业务函数只认这个接口和三种产品的接口，从头到尾不出现 Alipay、Stripe 这些具体类名。选哪个渠道，在入口选一次 gateway 就定下来了。

抽象工厂的结构：Client 只认 AbstractFactory 和抽象产品，每个 ConcreteFactory 产出一个系列

图里是 GoF 的角色，下表对应到本章的类。

角色	在本章案例里是谁	职责
抽象工厂（AbstractFactory）	`PaymentGateway`	声明一组创建方法，每种产品一个
具体工厂（ConcreteFactory）	`AlipayGateway`、`WechatGateway`、`StripeGateway`	实现全部创建方法，产出同一个渠道的一整套产品
抽象产品（AbstractProduct）	`PaymentClient`、`RefundClient`、`WebhookVerifier`	声明一种产品的接口
具体产品（ConcreteProduct）	`AlipayPaymentClient`、`WechatRefundClient`、`StripeWebhookVerifier` 等九个类	实现产品接口，属于某一个渠道
客户（Client）	`checkout`、`refund_order`、`handle_webhook`	只通过抽象工厂和抽象产品的接口干活

把具体工厂里的每个创建方法单独拿出来看，create_refund() 就是一个工厂方法：由子类决定创建哪个类。所以也可以说，抽象工厂是把几个必须一起变的工厂方法装进了同一个对象。

常见场景：

跨平台 UI 套件。同一套界面代码在 Windows 上要用 Windows 风格的按钮、复选框、滚动条，在 macOS 上要整套换成 macOS 风格，不能混着来。这是 GoF 原书举的例子。
主题和皮肤。浅色主题、深色主题各产出一套按钮、输入框、对话框，4.9 的练习就是这个。
数据库驱动。每种数据库配一套 SQL 编译器、类型映射、连接对象，4.6 的两个例子都是这样。
多环境配套对象。测试环境用一个产出假支付、假退款、永远通过验签的 FakeGateway，生产环境用真的；多租户系统里，每个租户一个工厂，产出这个租户自己的存储、消息队列客户端。

4.6 真实系统里的例子

例一：SQLAlchemy 的 Dialect（Python）

SQLAlchemy 是 Python 里常用的数据库工具包。同一句 select(...)，连 SQLite 和连 PostgreSQL 生成的 SQL 并不一样。它为每种数据库写了一个方言（Dialect）类，比如 SQLiteDialect 和 PGDialect。每个方言用类属性登记一整套配套的类：statement_compiler 是语句编译器，type_compiler_cls 是类型编译器，另外还有建表语句编译器、标识符加引号的规则等。编译语句时，SQLAlchemy 调 dialect.statement_compiler(dialect, statement) 新建编译器；编译器遇到 CAST 这类需要写类型名的地方，再通过 self.dialect.type_compiler_instance 向同一个方言要类型名。语句编译器本身也各家不同，比如只给 OFFSET、不给 LIMIT 时：SQLite 的语法不允许 OFFSET 单独出现，必须先写 LIMIT，于是补上表示不限行数的 LIMIT -1；PostgreSQL 则补上同样表示不限行数的 LIMIT ALL。简化版只保留这两个产品：

class TypeCompiler:  # 产品一：把通用类型翻译成本数据库的类型名（真实基类是 GenericTypeCompiler）
    names: dict[str, str]

    def __init__(self, dialect: "Dialect") -> None:
        self.dialect = dialect

    def process(self, type_: str) -> str:  # 真实版按类型对象分派到 visit_DATETIME 等方法
        return self.names[type_]


class SQLCompiler:  # 产品二：把语句编译成本数据库的 SQL（真实构造参数还有要编译的语句）
    offset_only: str  # 只给 OFFSET、不给 LIMIT 时要补的写法，各家不同（真实方法是 limit_clause）

    def __init__(self, dialect: "Dialect") -> None:
        self.dialect = dialect

    def compile(self, column: str, type_: str, offset: int) -> str:
        cast = self.dialect.type_compiler_instance.process(type_)  # 类型名向同一个方言要
        return f"SELECT CAST({column} AS {cast}) FROM events {self.offset_only} OFFSET {offset}"


class SQLiteTypeCompiler(TypeCompiler):
    names = {"DateTime": "DATETIME"}


class PGTypeCompiler(TypeCompiler):
    names = {"DateTime": "TIMESTAMP WITHOUT TIME ZONE"}


class SQLiteCompiler(SQLCompiler):
    offset_only = "LIMIT -1"


class PGCompiler(SQLCompiler):
    offset_only = "LIMIT ALL"


class Dialect:  # 抽象工厂（真实源码里这段 __init__ 在 DefaultDialect）
    name: str
    statement_compiler: type[SQLCompiler]
    type_compiler_cls: type[TypeCompiler]  # 1.x 里叫 type_compiler

    def __init__(self) -> None:
        self.type_compiler_instance = self.type_compiler_cls(self)


class SQLiteDialect(Dialect):
    name = "sqlite"
    statement_compiler = SQLiteCompiler
    type_compiler_cls = SQLiteTypeCompiler


class PGDialect(Dialect):
    name = "postgresql"
    statement_compiler = PGCompiler
    type_compiler_cls = PGTypeCompiler


for dialect in (SQLiteDialect(), PGDialect()):
    compiler = dialect.statement_compiler(dialect)
    print(f"{dialect.name}: {compiler.compile('created_at', 'DateTime', 20)}")

# 输出：
# sqlite: SELECT CAST(created_at AS DATETIME) FROM events LIMIT -1 OFFSET 20
# postgresql: SELECT CAST(created_at AS TIMESTAMP WITHOUT TIME ZONE) FROM events LIMIT ALL OFFSET 20

Dialect 对应本章的 PaymentGateway，SQLiteDialect、PGDialect 对应各个渠道的 gateway，两种编译器对应三件套。配套在这里同样要紧：PostgreSQL 的语句编译器要是拿到了 SQLite 的类型编译器，就会生成 CAST(created_at AS DATETIME)，PostgreSQL 执行时报 type "datetime" does not exist。SQLAlchemy 让编译器只能经由自己的方言拿类型编译器，这种组合就写不出来。和本章不同的是，它没有给每个产品写一个 create_xxx() 方法，而是把类本身存成类属性，用的时候直接调用类，原理和第 3 章 3.6 的映射表一样。

例二：TypeORM 的 Driver（TypeScript）

TypeORM 是 Node.js 生态里常用的 ORM，同时支持 PostgreSQL、MySQL 这类关系数据库和 MongoDB。它的源码里有一个 Driver 接口，每种数据库实现一个，比如 PostgresDriver、MongoDriver。Driver 里有两个创建方法：createQueryRunner() 返回执行语句的 QueryRunner，createSchemaBuilder() 返回按实体同步表结构的 SchemaBuilder。应用启动时调用 dataSource.synchronize()，它向驱动要一个 SchemaBuilder 并调用 build()，而 SchemaBuilder 干活时又经由同一个驱动拿 QueryRunner。简化版：

interface QueryRunner { // 产品一：执行语句（真实接口有几十个方法，都是异步的）
  query(sql: string): void;
}

interface SchemaBuilder { // 产品二：按实体同步表结构
  build(): void;
}

interface Driver { // 抽象工厂（真实 createQueryRunner 还有一个主从模式参数）
  createQueryRunner(): QueryRunner;
  createSchemaBuilder(): SchemaBuilder;
}

class PostgresQueryRunner implements QueryRunner {
  query(sql: string): void { console.log(`postgres 执行：${sql}`); }
}

class MongoQueryRunner implements QueryRunner {
  query(sql: string): void { throw new Error("Executing SQL query is not supported by MongoDB driver."); }
  createCollectionIndex(collection: string, field: string): void {
    console.log(`mongodb 建索引：${collection}.${field}`);
  }
}

class RdbmsSchemaBuilder implements SchemaBuilder { // 真实构造参数是 DataSource，经它向驱动要 QueryRunner
  constructor(private readonly driver: Driver) {}
  build(): void { this.driver.createQueryRunner().query('CREATE TABLE "user" ("email" varchar)'); }
}

class MongoSchemaBuilder implements SchemaBuilder {
  constructor(private readonly driver: Driver) {}
  build(): void {
    const runner = this.driver.createQueryRunner() as MongoQueryRunner; // 真实源码也这样转型
    runner.createCollectionIndex("user", "email");
  }
}

class PostgresDriver implements Driver {
  createQueryRunner(): QueryRunner { return new PostgresQueryRunner(); }
  createSchemaBuilder(): SchemaBuilder { return new RdbmsSchemaBuilder(this); }
}

class MongoDriver implements Driver {
  createQueryRunner(): QueryRunner { return new MongoQueryRunner(); }
  createSchemaBuilder(): SchemaBuilder { return new MongoSchemaBuilder(this); }
}

for (const driver of [new PostgresDriver(), new MongoDriver()]) {
  driver.createSchemaBuilder().build(); // 相当于 dataSource.synchronize()
}

// 输出：
// postgres 执行：CREATE TABLE "user" ("email" varchar)
// mongodb 建索引：user.email

Driver 对应 PaymentGateway，PostgresDriver、MongoDriver 对应各个渠道的 gateway。注意 MongoSchemaBuilder 里那个 as MongoQueryRunner：它敢把接口类型直接转成具体类，是因为它和这个 QueryRunner 出自同一个驱动，配套由构造保证。反过来，要是关系数据库的 RdbmsSchemaBuilder 配上了 MongoQueryRunner，同步表结构时第一次调用它就会抛出 MongoDB 驱动“不支持”的异常：真实源码里 MongoQueryRunner 的建表、查表结构这类方法都直接抛错，执行 SQL 的 query() 也一样（简化版只模拟了 query()）。

4.7 什么时候别用

产品族里只有一种产品。如果每个渠道只需要一个 PaymentClient，PaymentGateway 就只剩一个创建方法，那其实是第 3 章的工厂方法，或者一张“渠道名到类”的映射表。硬套抽象工厂，每个渠道要多写一个只有一行 return 的 gateway 类，读代码的人要多跳一层，换不来任何配套上的好处。
要频繁加新的产品种类。抽象工厂让“加一个渠道”很省事，代价是“加一种产品”很费事。比如财务要加对账功能，每个渠道再配一个 ReconcileClient：PaymentGateway 要加 create_reconcile()，三个 gateway 都要跟着改，有几个渠道就改几个类。更麻烦的是有的渠道不支持这个功能，只能在它的 gateway 里抛“不支持”的异常，接口上写着能造，调用时却会失败。如果你的系统里新功能比新渠道来得勤，先别急着建工厂接口。
配套关系还没稳定。只接了一个渠道时，你并不知道“一个渠道该有哪几件”。等第二个渠道接进来，常常发现它多了一件（比如需要先换取访问令牌的客户端）或者少了一件。先按 4.2 那样写，接第二、第三个渠道时再把真正配套的几件提成工厂接口，改动范围就是 4.4 这么大。

4.8 容易混淆的模式

模式	意图	区别
工厂方法（Factory Method）	父类定流程，子类决定创建哪种产品	一个工厂方法只产一种产品，靠继承扩展；抽象工厂里的每个创建方法，往往就是一个工厂方法
抽象工厂（Abstract Factory）	成套创建一组相互配合的产品	一个工厂对象有好几个创建方法，产出的对象必须属于同一族；它保证的是“配套”，不关心每个对象怎么一步步组装
建造者（Builder）	分步骤组装一个参数很多的对象	产出的是一个复杂对象，调用方一步步设置，最后取结果；抽象工厂是每调用一次就立刻返回一个完整的产品，见第 5 章

一句话区分：工厂方法造一个，抽象工厂造一套，建造者分好几步造一个复杂的。

4.9 练一练

后台管理系统要支持浅色和深色两种主题。界面上用到三种组件：按钮 Button、输入框 Input、对话框 Dialog，每种组件都有一个 render 方法，返回一段描述自己样子的字符串。同一个页面里，三种组件必须属于同一个主题，不能出现浅色输入框配深色按钮。

请用抽象工厂实现：

抽象工厂 ThemeFactory 有三个创建方法：create_button()、create_input()、create_dialog()。
具体工厂 LightTheme 和 DarkTheme 各产出自己的一套组件，比如 LightButton 渲染成 [浅色按钮白底黑字] 登录。
写一个 render_login_form(theme)，用传进来的主题画出登录表单：一个“请输入邮箱”的输入框、一个“登录”按钮、一个“提示：密码错误”的对话框。分别用两种主题各调用一次，Python 和 TS 各写一版。

提示：render_login_form 里不应该出现 Light 或 Dark 这样的字样。做完想一想：再加一个“高对比度”主题，你要动哪几个地方？如果要加第四种组件“下拉框”呢？

4.10 小结

抽象工厂用一个工厂对象成套地创建一组配合使用的对象。业务代码只拿一个工厂，需要哪个对象就向它要，换一个工厂就换掉整套。
它最大的好处是配套由构造保证：同一个工厂产出的对象不会来自两个族，一个族用哪几个类只写在一个具体工厂里，漏实现的创建方法在第一次创建这个工厂（Python）或编译（TS）时就会报错。
加一个新的族（新渠道、新主题）就是加一个具体工厂，业务代码不动；入口的映射表照样要加一行登记。
加一种新产品要改抽象工厂接口和所有具体工厂。产品种类经常变时，它会成为负担。
产品族里只有一种产品时，用第 3 章的工厂方法或映射表就够了。