跳转到内容

第 3 章 · 工厂方法(Factory Method)

一句话:父类写好“拿一个对象来做事”的流程,“创建哪一种对象”这一步留给子类决定。 解决的痛:同一串按类型 new 对象的 if 散在好几个调用点,每加一种类型就要把每一处都改一遍。

你在一家做在线课程的公司写后台。用户注册成功、申请重置密码时,系统都要给用户发一条通知。用户可以在个人设置里选通知渠道,目前支持两种:邮件和短信。

发邮件和发短信的细节已经有人封装好了:EmailSender 和 SmsSender 两个类,各有一个 send(to, message) 方法。真实项目里它们背后是邮件服务和短信服务商的 SDK,本章用 print 代替。

你要写的是两个业务函数:注册成功时调用的 on_registered,密码重置时调用的 on_password_reset。它们都要做三件事:把文案套进统一的模板(开头加”【通知】“和用户名),按用户选的渠道发出去,再记一条日志,方便客服查”这条通知到底发了没有”。

最直接的写法是在每个业务函数里判断渠道,是邮件就 new 一个 EmailSender,是短信就 new 一个 SmsSender:

class EmailSender:
def send(self, to: str, message: str) -> None:
print(f"[邮件] 发给 {to}:{message}")
class SmsSender:
def send(self, to: str, message: str) -> None:
print(f"[短信] 发给 {to}:{message}")
# 调用点一:注册成功
def on_registered(channel: str, user: str) -> None:
message = f"【通知】{user},欢迎注册!"
if channel == "email":
EmailSender().send(user, message)
elif channel == "sms":
SmsSender().send(user, message)
else:
raise ValueError(f"未知渠道:{channel}")
print(f"[日志] 已通过 {channel} 通知 {user}")
# 调用点二:密码重置
def on_password_reset(channel: str, user: str, code: str) -> None:
message = f"【通知】{user},你的验证码是 {code},10 分钟内有效。"
if channel == "email":
EmailSender().send(user, message)
elif channel == "sms":
SmsSender().send(user, message)
else:
raise ValueError(f"未知渠道:{channel}")
print(f"[日志] 已通过 {channel} 通知 {user}")
if __name__ == "__main__":
on_registered("email", "小明")
on_password_reset("sms", "小红", "4821")
# 输出:
# [邮件] 发给 小明:【通知】小明,欢迎注册!
# [日志] 已通过 email 通知 小明
# [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
# [日志] 已通过 sms 通知 小红

TypeScript 版结构一样:

class EmailSender {
send(to: string, message: string): void {
console.log(`[邮件] 发给 ${to}:${message}`);
}
}
class SmsSender {
send(to: string, message: string): void {
console.log(`[短信] 发给 ${to}:${message}`);
}
}
// 调用点一:注册成功
function onRegistered(channel: string, user: string): void {
const message = `【通知】${user},欢迎注册!`;
if (channel === "email") {
new EmailSender().send(user, message);
} else if (channel === "sms") {
new SmsSender().send(user, message);
} else {
throw new Error(`未知渠道:${channel}`);
}
console.log(`[日志] 已通过 ${channel} 通知 ${user}`);
}
// 调用点二:密码重置
function onPasswordReset(channel: string, user: string, code: string): void {
const message = `【通知】${user},你的验证码是 ${code},10 分钟内有效。`;
if (channel === "email") {
new EmailSender().send(user, message);
} else if (channel === "sms") {
new SmsSender().send(user, message);
} else {
throw new Error(`未知渠道:${channel}`);
}
console.log(`[日志] 已通过 ${channel} 通知 ${user}`);
}
onRegistered("email", "小明");
onPasswordReset("sms", "小红", "4821");
// 输出:
// [邮件] 发给 小明:【通知】小明,欢迎注册!
// [日志] 已通过 email 通知 小明
// [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
// [日志] 已通过 sms 通知 小红

两个函数各自完整,从上往下读就知道一条通知是怎么发出去的,出了问题也好排查。大多数人的第一版都这么写,渠道只有两种、发通知的业务只有两处的时候,它完全够用。

产品经理提了两条:加一个微信渠道,用公众号模板消息发;短信在发送前要校验长度,超过 30 字的不发,直接报错。后一条是因为短信按条计费,超长的会被拆成两条收钱,运营要求先把文案改短。

要改的地方有五处:

  1. 新增 WechatSender 类。这一处任何写法都躲不掉。
  2. on_registered:在 if 里加一个 elif channel == "wechat" 分支。
  3. on_password_reset:把同样的分支再加一遍。
  4. on_registered 的短信分支:在 SmsSender().send 前面加长度校验。
  5. on_password_reset 的短信分支:把这段校验复制过去。

改完之后,每个业务函数都长这样:

def on_registered(channel: str, user: str) -> None:
message = f"【通知】{user},欢迎注册!"
if channel == "email":
EmailSender().send(user, message)
elif channel == "sms":
if len(message) > 30: # 新增:短信长度校验
raise ValueError(f"短信超过 30 字({len(message)} 字),未发送")
SmsSender().send(user, message)
elif channel == "wechat": # 新增:微信分支
WechatSender().send(user, message)
...
# on_password_reset:上面两段新增原样再抄一遍

TS 版要改的是同样的五处:新增 WechatSender,onRegistered 和 onPasswordReset 各加一个 channel === "wechat" 分支,各在短信分支里加一段长度校验。

最容易出事的是第 5 处。30 这个数字现在写了两份,下个月运营把上限放宽到 40 字,负责改的人搜到注册那一处改了,密码重置那一处还是 30。一条 35 字的验证码短信会在密码重置时被拦下,用户收不到验证码,而注册通知一切正常,测试只测了注册的话发现不了。何况真实项目里发通知的业务不止两处,下单成功、发货、退款各有一份同样的 if,这五处改动很快会变成十几处。

测试也跟着麻烦。写 on_password_reset 的单元测试时,你不想真的发短信,想换成一个只把消息记下来的假渠道。可是创建哪个 Sender 写死在函数里面:要么在两个 if 里各加一个 elif channel == "fake",让测试用的分支进入生产代码;要么用 mock 去替换模块里的 SmsSender 这个名字,测试就得知道业务函数内部 new 了哪个类。

问题的形状是:每加一种渠道、每加一条渠道自己的规则,就要把每个调用点的 if 改一遍;调用点有几个,就改几遍。

朴素写法:五处要改的位置

图中 ①–⑤ 就是上面列出的五处,两个业务函数各自连着所有渠道,规则也各抄一份。

最先想到的改法,是把“按渠道 new 对象”这件事收进一个函数 create_sender(channel),两个业务函数都调它。这种写法通常叫简单工厂(Simple Factory)。它不在 GoF 的 23 个模式里,但非常常用:

from typing import Protocol
SMS_MAX_LENGTH = 30
class Sender(Protocol):
def send(self, to: str, message: str) -> None: ...
class EmailSender:
def send(self, to: str, message: str) -> None:
print(f"[邮件] 发给 {to}:{message}")
class SmsSender:
def send(self, to: str, message: str) -> None:
print(f"[短信] 发给 {to}:{message}")
class WechatSender:
def send(self, to: str, message: str) -> None:
print(f"[微信] 发给 {to}:{message}")
def create_sender(channel: str) -> Sender: # 简单工厂:所有 new 集中在这一个函数
if channel == "email":
return EmailSender()
if channel == "sms":
return SmsSender()
if channel == "wechat":
return WechatSender()
raise ValueError(f"未知渠道:{channel}")
def check_sms(message: str) -> None:
if len(message) > SMS_MAX_LENGTH:
raise ValueError(f"短信超过 {SMS_MAX_LENGTH} 字({len(message)} 字),未发送")
# 调用点一:注册成功
def on_registered(channel: str, user: str) -> None:
message = f"【通知】{user},欢迎注册!"
if channel == "sms": # 校验没有地方放,还是留在调用点
check_sms(message)
create_sender(channel).send(user, message)
print(f"[日志] 已通过 {channel} 通知 {user}")
# 调用点二:密码重置
def on_password_reset(channel: str, user: str, code: str) -> None:
message = f"【通知】{user},你的验证码是 {code},10 分钟内有效。"
if channel == "sms": # 同一段判断,又写了一遍
check_sms(message)
create_sender(channel).send(user, message)
print(f"[日志] 已通过 {channel} 通知 {user}")
if __name__ == "__main__":
on_registered("wechat", "小刚")
on_password_reset("sms", "小红", "4821")
# 输出:
# [微信] 发给 小刚:【通知】小刚,欢迎注册!
# [日志] 已通过 wechat 通知 小刚
# [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
# [日志] 已通过 sms 通知 小红
const SMS_MAX_LENGTH = 30;
interface Sender {
send(to: string, message: string): void;
}
class EmailSender implements Sender {
send(to: string, message: string): void {
console.log(`[邮件] 发给 ${to}:${message}`);
}
}
class SmsSender implements Sender {
send(to: string, message: string): void {
console.log(`[短信] 发给 ${to}:${message}`);
}
}
class WechatSender implements Sender {
send(to: string, message: string): void {
console.log(`[微信] 发给 ${to}:${message}`);
}
}
function createSender(channel: string): Sender { // 简单工厂:所有 new 集中在这一个函数
if (channel === "email") return new EmailSender();
if (channel === "sms") return new SmsSender();
if (channel === "wechat") return new WechatSender();
throw new Error(`未知渠道:${channel}`);
}
function checkSms(message: string): void {
if (message.length > SMS_MAX_LENGTH) {
throw new Error(`短信超过 ${SMS_MAX_LENGTH} 字(${message.length} 字),未发送`);
}
}
// 调用点一:注册成功
function onRegistered(channel: string, user: string): void {
const message = `【通知】${user},欢迎注册!`;
if (channel === "sms") checkSms(message); // 校验没有地方放,还是留在调用点
createSender(channel).send(user, message);
console.log(`[日志] 已通过 ${channel} 通知 ${user}`);
}
// 调用点二:密码重置
function onPasswordReset(channel: string, user: string, code: string): void {
const message = `【通知】${user},你的验证码是 ${code},10 分钟内有效。`;
if (channel === "sms") checkSms(message); // 同一段判断,又写了一遍
createSender(channel).send(user, message);
console.log(`[日志] 已通过 ${channel} 通知 ${user}`);
}
onRegistered("wechat", "小刚");
onPasswordReset("sms", "小红", "4821");
// 输出:
// [微信] 发给 小刚:【通知】小刚,欢迎注册!
// [日志] 已通过 wechat 通知 小刚
// [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
// [日志] 已通过 sms 通知 小红

简单工厂解决了 3.3 里的第 2、3 处:选哪个 Sender 只在 create_sender 里判断一次,加微信时两个业务函数都不用动。

它没解决的是渠道自己的规则。加渠道要在 create_sender 里加一个分支,这本身不算麻烦,后面会看到,任何写法都得在某个地方登记一下新渠道。真正的问题是短信校验没有地方放:create_sender 只负责“造出来”,管不了“发之前检查什么”,于是 if channel == "sms" 又回到了每个调用点,第 4、5 处的复制一处没少。你也许会想把校验写进 SmsSender.send,可 Sender 通常只是服务商 SDK 的一层薄包装,“文案不超过 30 字”是我们自己的运营规则;放进去以后,别的业务想直接用 SmsSender 发一条长短信就发不了了。

你还可能想再往前一步,写一个 send_notification(channel, user, text),把模板、校验、创建、日志全收进去。这样调用点确实只剩一行,但这个函数里会有两串按渠道的判断:一串在 create_sender 里决定 new 谁,一串决定校验什么。同一个渠道的知识分在两串 if 里,加一个带自己规则的渠道,两串都要改,漏改一串也不报错。

工厂方法的做法是:把“发一条通知”的公共流程写进基类 Notifier 的 notify 方法,流程里需要一个 Sender 时,调用自己的 create_sender() 去要。基类不实现这个方法,由每个渠道的子类实现。和这个渠道有关的规则(比如短信的长度校验)也写在对应的子类里:

from abc import ABC, abstractmethod
from typing import Protocol
class Sender(Protocol): # 产品:只管把消息发出去
def send(self, to: str, message: str) -> None: ...
class EmailSender:
def send(self, to: str, message: str) -> None:
print(f"[邮件] 发给 {to}:{message}")
class SmsSender:
def send(self, to: str, message: str) -> None:
print(f"[短信] 发给 {to}:{message}")
class WechatSender:
def send(self, to: str, message: str) -> None:
print(f"[微信] 发给 {to}:{message}")
class Notifier(ABC): # 创建者:公共流程写在这里
channel: str
@abstractmethod
def create_sender(self) -> Sender: # 工厂方法:创建哪种 Sender 由子类决定
...
def check(self, message: str) -> None: # 发送前校验,默认什么都不做
pass
def notify(self, user: str, text: str) -> None:
message = f"【通知】{user},{text}"
self.check(message)
self.create_sender().send(user, message)
print(f"[日志] 已通过{self.channel}通知 {user}")
class EmailNotifier(Notifier):
channel = "邮件"
def create_sender(self) -> Sender:
return EmailSender()
class SmsNotifier(Notifier):
channel = "短信"
max_length = 30
def create_sender(self) -> Sender:
return SmsSender()
def check(self, message: str) -> None: # 短信的规则只写在这一处
if len(message) > self.max_length:
raise ValueError(f"短信超过 {self.max_length} 字({len(message)} 字),未发送")
class WechatNotifier(Notifier): # 新渠道:加这一个类,别的都不动
channel = "微信"
def create_sender(self) -> Sender:
return WechatSender()
# 调用点一:注册成功
def on_registered(notifier: Notifier, user: str) -> None:
notifier.notify(user, "欢迎注册!")
# 调用点二:密码重置
def on_password_reset(notifier: Notifier, user: str, code: str) -> None:
notifier.notify(user, f"你的验证码是 {code},10 分钟内有效。")
if __name__ == "__main__":
on_registered(EmailNotifier(), "小明")
on_password_reset(SmsNotifier(), "小红", "4821")
on_registered(WechatNotifier(), "小刚")
try:
SmsNotifier().notify("小红", "你的订单已发货,快递单号 SF1234567890,预计明天送达。")
except ValueError as error:
print(f"发送失败:{error}")
# 输出:
# [邮件] 发给 小明:【通知】小明,欢迎注册!
# [日志] 已通过邮件通知 小明
# [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
# [日志] 已通过短信通知 小红
# [微信] 发给 小刚:【通知】小刚,欢迎注册!
# [日志] 已通过微信通知 小刚
# 发送失败:短信超过 30 字(40 字),未发送

TS 版用 interface 表示产品,用抽象类表示创建者:

interface Sender { // 产品:只管把消息发出去
send(to: string, message: string): void;
}
class EmailSender implements Sender {
send(to: string, message: string): void {
console.log(`[邮件] 发给 ${to}:${message}`);
}
}
class SmsSender implements Sender {
send(to: string, message: string): void {
console.log(`[短信] 发给 ${to}:${message}`);
}
}
class WechatSender implements Sender {
send(to: string, message: string): void {
console.log(`[微信] 发给 ${to}:${message}`);
}
}
abstract class Notifier { // 创建者:公共流程写在这里
abstract readonly channel: string;
abstract createSender(): Sender; // 工厂方法:创建哪种 Sender 由子类决定
check(message: string): void {} // 发送前校验,默认什么都不做
notify(user: string, text: string): void {
const message = `【通知】${user},${text}`;
this.check(message);
this.createSender().send(user, message);
console.log(`[日志] 已通过${this.channel}通知 ${user}`);
}
}
class EmailNotifier extends Notifier {
readonly channel = "邮件";
createSender(): Sender {
return new EmailSender();
}
}
class SmsNotifier extends Notifier {
readonly channel = "短信";
readonly maxLength = 30;
createSender(): Sender {
return new SmsSender();
}
check(message: string): void { // 短信的规则只写在这一处
if (message.length > this.maxLength) {
throw new Error(`短信超过 ${this.maxLength} 字(${message.length} 字),未发送`);
}
}
}
class WechatNotifier extends Notifier { // 新渠道:加这一个类,别的都不动
readonly channel = "微信";
createSender(): Sender {
return new WechatSender();
}
}
// 调用点一:注册成功
function onRegistered(notifier: Notifier, user: string): void {
notifier.notify(user, "欢迎注册!");
}
// 调用点二:密码重置
function onPasswordReset(notifier: Notifier, user: string, code: string): void {
notifier.notify(user, `你的验证码是 ${code},10 分钟内有效。`);
}
onRegistered(new EmailNotifier(), "小明");
onPasswordReset(new SmsNotifier(), "小红", "4821");
onRegistered(new WechatNotifier(), "小刚");
try {
new SmsNotifier().notify("小红", "你的订单已发货,快递单号 SF1234567890,预计明天送达。");
} catch (error) {
console.log(`发送失败:${(error as Error).message}`);
}
// 输出:
// [邮件] 发给 小明:【通知】小明,欢迎注册!
// [日志] 已通过邮件通知 小明
// [短信] 发给 小红:【通知】小红,你的验证码是 4821,10 分钟内有效。
// [日志] 已通过短信通知 小红
// [微信] 发给 小刚:【通知】小刚,欢迎注册!
// [日志] 已通过微信通知 小刚
// 发送失败:短信超过 30 字(40 字),未发送

工厂方法版:Notifier 与 Sender 两条继承线

看标红的 WechatNotifier 和 WechatSender:新渠道就是这一对新类。

一次短信通知的调用顺序

看顺序图:短信长度校验发生在 SmsNotifier 内部,先于 SmsSender.send。

逐条对照 3.3 的五处改动:

  • 第 1 处,新增 WechatSender:还是要写,这是新渠道本身的代码。
  • 第 2、3 处,两个业务函数里的微信分支:不用加了。业务函数只收一个 Notifier,里面没有任何按渠道的判断。加微信是新增 WechatNotifier 这一个类,Notifier、另外两个子类和两个业务函数一行没动;程序入口把字符串换成对象的映射表加一行(见本节末尾)。和简单工厂比,差别不在少改这一行,而在于微信的一切(用哪个 Sender、有什么规则)都在 WechatNotifier 一个类里,不会分散到两串 if 里。
  • 第 4、5 处,短信校验:只写在 SmsNotifier.check 里,只有一份。运营把上限改成 40 字,改 max_length 这一行,注册、密码重置以及以后所有发短信的业务都跟着变,不会再漏掉哪一处。
  • 模板和日志:原来每个业务函数各写一遍,现在只在 Notifier.notify 里出现一次。日志里的渠道名改由每个子类的 channel 给出,所以打印的是“邮件”而不是 email。

测试时换假渠道,也不用碰生产代码,在测试文件里写一个子类就行:

class FakeSender:
def __init__(self) -> None:
self.sent: list[str] = []
def send(self, to: str, message: str) -> None:
self.sent.append(message)
class FakeNotifier(Notifier):
channel = "测试渠道"
def __init__(self) -> None:
self.sender = FakeSender()
def create_sender(self) -> Sender:
return self.sender
# 调用 on_password_reset(notifier, "小红", "4821") 之后,检查 notifier.sender.sent

有一处判断并没有消失:用户在设置里选的是字符串 "sms",总得有一个地方把它变成 SmsNotifier()。区别在于这个判断只在程序入口出现一次,业务函数拿到的已经是对象;3.6 会给出它最省事的写法。代价是类变多了:原来两个函数,现在一个基类加三个子类。

GoF 对工厂方法的定义是:定义一个创建对象的接口,让子类决定实例化哪个类。工厂方法让一个类把实例化推迟到它的子类。

换成本章的话说:Notifier 知道发通知的每一步,但不知道、也不想知道用哪个 Sender 发。它只声明“我需要一个 Sender”,也就是 create_sender(),具体给哪一个,由 EmailNotifier、SmsNotifier 这些子类回答。notify 里调用 create_sender() 的那一行,就是父类把“创建什么”让给子类的地方。

工厂方法的结构:Creator 的 anOperation 调用 factoryMethod,由 ConcreteCreator 决定创建哪种 Product

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

角色 在本章案例里是谁 职责
产品(Product) Sender 声明所有产品共有的接口,这里是 send
具体产品(ConcreteProduct) EmailSender、SmsSender、WechatSender 实现产品接口,真正把消息发出去
创建者(Creator) Notifier 声明工厂方法 create_sender,在 notify 里用它返回的产品走完公共流程
具体创建者(ConcreteCreator) EmailNotifier、SmsNotifier、WechatNotifier 实现工厂方法,返回对应的具体产品;渠道自己的规则也放在这里

注意创建者的主要工作并不是“创建”,而是那段公共流程,创建只是流程里留给子类填的一步。父类定流程、子类填其中一步,这本身是模板方法(Template Method,第 20 章)模式的形状;工厂方法可以看成它的一个特例:子类要填的那一步,恰好是创建一个对象。

常见场景:

  • 框架给使用者留的钩子。框架写好流程,使用者通过继承决定流程里用哪个对象。比如 Django 的类视图里,get_form() 负责创建表单,子类覆盖它就能换一种表单,视图的其余流程不变。
  • 按配置选实现。日志写到文件还是控制台,缓存放内存还是 Redis,同一套流程配上不同的具体创建者。
  • 测试替身。像 3.4 的 FakeNotifier 那样,测试里写一个子类返回假对象,生产代码不用为测试改一行。
  • 两组一一对应的类。每个具体创建者正好对应一个具体产品,比如每种浏览器启动器对应一种浏览器,3.7 的例二就是这样。

在 Python 里,类本身就是可调用的对象,EmailSender() 和调用一个返回 EmailSender 实例的函数没有区别。所以简单工厂不需要一串 if,一张从渠道名到类的表就够了:

from collections.abc import Callable
from typing import Protocol
class Sender(Protocol):
def send(self, to: str, message: str) -> None: ...
class EmailSender:
def send(self, to: str, message: str) -> None:
print(f"[邮件] 发给 {to}:{message}")
class SmsSender:
def send(self, to: str, message: str) -> None:
print(f"[短信] 发给 {to}:{message}")
SENDERS: dict[str, Callable[[], Sender]] = {"email": EmailSender, "sms": SmsSender}
def create_sender(channel: str) -> Sender:
if channel not in SENDERS:
raise ValueError(f"未知渠道:{channel}")
return SENDERS[channel]()
if __name__ == "__main__":
create_sender("email").send("小明", "欢迎注册!")
create_sender("sms").send("小红", "你的验证码是 4821")
# 输出:
# [邮件] 发给 小明:欢迎注册!
# [短信] 发给 小红:你的验证码是 4821

TS 里类也可以当值传来传去,映射表的值类型写成构造签名 new () => Sender:

interface Sender {
send(to: string, message: string): void;
}
class EmailSender implements Sender {
send(to: string, message: string): void {
console.log(`[邮件] 发给 ${to}:${message}`);
}
}
class SmsSender implements Sender {
send(to: string, message: string): void {
console.log(`[短信] 发给 ${to}:${message}`);
}
}
const SENDERS: Record<string, new () => Sender> = { email: EmailSender, sms: SmsSender };
function createSender(channel: string): Sender {
const SenderClass = SENDERS[channel];
if (SenderClass === undefined) throw new Error(`未知渠道:${channel}`);
return new SenderClass();
}
createSender("email").send("小明", "欢迎注册!");
createSender("sms").send("小红", "你的验证码是 4821");
// 输出:
// [邮件] 发给 小明:欢迎注册!
// [短信] 发给 小红:你的验证码是 4821

加微信只是在表里加一行 "wechat": WechatSender,create_sender 本身不用改。结构还是同一个:“按名字找到要创建的类,再创建它”。语言替你省掉的是那串 if,以及一个个只有一行 return XxxSender() 的创建者子类:子类唯一的工作就是返回某个类的实例时,它和表里的一行没有区别。

所以判断标准是:创建的旁边没有公共流程时,这张表就够了。只是“按配置拿到一个对象,然后调用它的方法”,用表;创建前后还有一段每个渠道都要走的流程(模板、校验、日志),而且有的渠道要往流程里插入自己的规则,工厂方法的基类才有东西可装,子类也才不只是一行。3.4 结尾说的“程序入口那一处判断”,就适合写成 NOTIFIERS = {"email": EmailNotifier, "sms": SmsNotifier, "wechat": WechatNotifier},两种写法常常这样搭配使用。

例一:Python 标准库 http.server(Python)

Section titled “例一:Python 标准库 http.server(Python)”

用标准库写一个最小的 Web 服务,只要写一个处理请求的类,再把这个类交给服务器:HTTPServer(("", 8000), MyHandler)。注意传进去的是类,不是对象。服务器每收到一个请求,就用这个类新建一个处理器对象来处理它。这段逻辑在 HTTPServer 的祖先类 socketserver.BaseServer 里:process_request 是公共流程(处理请求、关闭连接),其中的 finish_request 只有一行 self.RequestHandlerClass(request, client_address, self)。下面是去掉网络部分的简化版:

from __future__ import annotations
class BaseRequestHandler: # 产品:每个请求一个
def __init__(self, request: str, client_address: tuple[str, int], server: BaseServer) -> None:
self.request = request
self.client_address = client_address
self.server = server
self.handle()
def handle(self) -> None:
pass
class BaseServer: # 创建者:接请求、建处理器、关连接的流程写在这里
def __init__(self, server_address: tuple[str, int],
RequestHandlerClass: type[BaseRequestHandler]) -> None:
self.server_address = server_address
self.RequestHandlerClass = RequestHandlerClass
def process_request(self, request: str, client_address: tuple[str, int]) -> None:
self.finish_request(request, client_address)
self.shutdown_request(request)
def finish_request(self, request: str, client_address: tuple[str, int]) -> None:
self.RequestHandlerClass(request, client_address, self) # 工厂方法就在这一行:new 哪个处理器
def shutdown_request(self, request: str) -> None:
print(f"关闭连接:{request}")
class HelloHandler(BaseRequestHandler): # 使用者只写这一个类
def handle(self) -> None:
print(f"{self.client_address[0]} 请求 {self.request},回复 Hello")
if __name__ == "__main__":
server = BaseServer(("127.0.0.1", 8000), HelloHandler)
server.process_request("GET /", ("10.0.0.5", 51234))
server.process_request("GET /about", ("10.0.0.6", 51235))
# 输出:
# 10.0.0.5 请求 GET /,回复 Hello
# 关闭连接:GET /
# 10.0.0.6 请求 GET /about,回复 Hello
# 关闭连接:GET /about

BaseServer 对应本章的 Notifier,process_request 对应 notify,finish_request 就是工厂方法,处理器类对应 Sender。不同的是,它没有让你写一个 HelloServer 子类去覆盖 finish_request,而是让你把类当参数传进来,正是 3.6 说的“类本身就是工厂”。需要时你仍然可以继承 BaseServer 覆盖 finish_request,两条路都留着。

例二:Playwright 的 BrowserType(TypeScript)

Section titled “例二:Playwright 的 BrowserType(TypeScript)”

Playwright 是一个浏览器自动化测试库。写测试时,你通过 chromium.launch() 或 firefox.launch() 拿到一个 Browser,后面开页面、点按钮的代码对两种浏览器完全一样。它的服务端源码里有一个抽象基类 BrowserType,“启动一个浏览器”的公共流程写在基类里,Chromium、Firefox、WebKit 各自继承它,补上和自己那种浏览器建立连接的部分。简化版只保留这个骨架:

interface Browser { // 产品:调用方只认这个接口
version(): string;
close(): Promise<void>;
}
class ChromiumBrowser implements Browser {
version(): string { return "Chromium(模拟)"; }
async close(): Promise<void> { console.log("Chromium 已关闭"); }
}
class FirefoxBrowser implements Browser {
version(): string { return "Firefox(模拟)"; }
async close(): Promise<void> { console.log("Firefox 已关闭"); }
}
abstract class BrowserType { // 创建者:launch 的公共流程写在这里
abstract name(): string;
// 工厂方法:连上哪种浏览器由子类决定(真实源码里是异步的 _connectToTransport)
protected abstract connect(): Browser;
async launch(): Promise<Browser> {
console.log(`启动 ${this.name()} 进程`);
const browser = this.connect();
console.log(`已连接 ${browser.version()}`);
return browser;
}
}
class Chromium extends BrowserType {
name(): string { return "chromium"; }
protected connect(): Browser { return new ChromiumBrowser(); }
}
class Firefox extends BrowserType {
name(): string { return "firefox"; }
protected connect(): Browser { return new FirefoxBrowser(); }
}
const chromium = new Chromium();
const firefox = new Firefox();
async function smokeTest(browserType: BrowserType): Promise<void> { // 调用方只认 BrowserType 和 Browser
const browser = await browserType.launch();
await browser.close();
}
async function main(): Promise<void> {
await smokeTest(chromium);
await smokeTest(firefox);
}
void main();
// 输出:
// 启动 chromium 进程
// 已连接 Chromium(模拟)
// Chromium 已关闭
// 启动 firefox 进程
// 已连接 Firefox(模拟)
// Firefox 已关闭

BrowserType 对应 Notifier,launch 对应 notify,connect 对应 create_sender,Browser 对应 Sender。同一个 smokeTest 换一个 BrowserType 就能在另一种浏览器上跑,就像 on_registered 换一个 Notifier 就换了渠道。

  • 只有一种产品,而且不会再变。比如系统只发邮件,也没有计划加别的渠道,直接在 notify 里写 EmailSender().send(...) 就好。套上工厂方法以后,读代码的人要从 Notifier 跳到 EmailNotifier 再跳到 EmailSender,看三个类才弄清“发一封邮件”这一件事。
  • 为了“以后可能要扩展”提前抽象。还没出现第二种渠道就先写好基类和子类,等第二种真的来了,常常发现它要的流程和你预想的不一样,比如微信要先查用户有没有关注公众号,原来的 notify 得重写,提前写的抽象反而要先拆掉。等第二种出现时再重构也不迟:到时只需把两个业务函数里的 if 提成一个基类和两个子类,改动范围就是本章 3.4 这么大。
  • 创建的旁边没有公共流程。只是按名字拿一个对象,就用 3.6 的映射表。为每个产品配一个只有一行的创建者子类,加一种产品就要加两个类,类的数量翻了一倍,却没换来任何好处。
模式 意图 区别
简单工厂(Simple Factory) 用一个函数按参数集中创建对象 不在 GoF 的 23 个模式里;没有继承,一个函数加一串 if 或一张表;只管创建,产品自己的规则(如短信校验)没有地方放,会回到每个调用点
工厂方法(Factory Method) 父类定流程,子类决定创建哪种产品 靠继承;每个具体创建者负责一种产品,和这种产品有关的规则也跟着放进子类;加产品是加一个子类,业务函数和其他子类不改,入口的映射表加一行
抽象工厂(Abstract Factory) 成套创建一组相互配合的产品 一个工厂有好几个创建方法,产出的对象必须配套,比如同一套皮肤的按钮和输入框;每个创建方法常常就是一个工厂方法;见第 4 章

一句话区分:简单工厂是一个函数,工厂方法是一个方法加一组子类,抽象工厂是一个对象负责一整套产品。

后台有一个“导出报表”功能,现在写成一个函数 export_report(fmt, data):先输出表头 === 报表 ===,再用 if fmt == "csv" 和 elif fmt == "json" 分别把数据转成 CSV 或 JSON,最后输出页脚 === 共 N 行 ===。另一个“导出订单”功能里,同样的表头、页脚和 if 又写了一遍。

请把它改成工厂方法:

  • 基类 ReportExporter 有一个 export(data) 方法,负责统一的表头和页脚,中间的正文交给 create_writer() 返回的对象去写。
  • 子类 CsvReportExporter 和 JsonReportExporter 各自实现 create_writer(),分别返回 CsvWriter 和 JsonWriter。
  • 用两行数据 [{"region": "华东", "sales": 120}, {"region": "华南", "sales": 95}] 分别导出一次,Python 和 TS 各写一版。

提示:先定义一个 Writer 接口,只有一个方法 write(rows),返回字符串。JSON 输出不要带空格,中文也不要转义成 \u 编码,这样 Python 的 json.dumps(要传 ensure_ascii=False 和紧凑的 separators)和 TS 的 JSON.stringify 结果才一样。做完想一想:以后要加 Excel 导出,你要动哪几个地方?

  • 工厂方法把“创建哪种对象”从公共流程里抽出来,变成一个由子类实现的方法:父类只管用,子类只管造。
  • 加一种产品就是加一个具体创建者子类,业务调用点和其他子类都不用改;和这种产品有关的规则也只放在它自己的子类里。
  • 简单工厂把散落的 new 收到一处,已经解决了一大半问题;但它只管创建,产品自己的规则(比如短信校验)没有地方放,又会回到每个调用点。
  • 在 Python 和 TS 里,类可以当值使用,一张“名字到类”的表就是最简洁的简单工厂。创建的旁边没有公共流程时用表,有公共流程才值得用工厂方法。
  • 只有一种产品,或者只是为了“以后可能扩展”时,不要用它。