跳转到内容

第 6 章 · 单例(Singleton)

一句话:让一个类自己保证只有一个实例,并给出一个所有代码都知道去哪儿拿它的入口。 解决的痛:创建一次很贵的对象(要联网、要建连接)被每个调用点各造一份,代价翻了好几倍,测试时还没法换成假的。

还是那家做在线课程的公司。后台有一份配置文件 settings.json,放着几个模块都要用的设置:下单用的货币、短信的字数上限、生成日报要连的数据库地址。程序启动以后,这几项基本不会变。

用到配置的有三个模块:orders.py 下单时要知道货币;notifications.py 发短信前要知道字数上限;reports.py 生成日报时要知道连哪个数据库。

你要做的是让这三个模块都拿得到配置。本章的代码用一个计数器代替真正的文件读取,每读一次打印一行。

最直接的写法是写一个 Config 类,构造时读文件;谁要用配置,谁就当场 Config() 一个。另一种同样常见的写法是在 config.py 里导入时就读好,放进一个模块级的全局变量 SETTINGS,别的模块 from config import SETTINGS。下面两种都有:订单和短信用前一种,日报用后一种。

read_count = 0
def read_config_file() -> dict[str, str]: # 真实项目里是 json.load(open("settings.json"))
global read_count
read_count += 1
print(f"读取 settings.json(第 {read_count} 次)")
return {"currency": "CNY", "sms_max_length": "30", "report_db": "postgres://report-db/courses"}
class Config:
def __init__(self) -> None:
self.values = read_config_file()
def get(self, key: str) -> str:
return self.values[key]
# config.py 里的另一种写法:导入时就读好,放进模块级全局变量
SETTINGS = Config()
# orders.py
def place_order(user: str, price: int) -> None:
currency = Config().get("currency")
print(f"{user} 下单:{price} {currency}")
# notifications.py
def send_sms(user: str, text: str) -> None:
limit = int(Config().get("sms_max_length"))
if len(text) > limit:
print(f"短信超过 {limit} 字,未发送")
return
print(f"短信发给 {user}:{text}")
# reports.py
def daily_report() -> None:
print(f"从 {SETTINGS.get('report_db')} 生成日报")
if __name__ == "__main__":
place_order("小明", 199)
place_order("小红", 299)
send_sms("小红", "你的验证码是 4821")
daily_report()
print(f"共读取配置 {read_count} 次")
# 输出:
# 读取 settings.json(第 1 次)
# 读取 settings.json(第 2 次)
# 小明 下单:199 CNY
# 读取 settings.json(第 3 次)
# 小红 下单:299 CNY
# 读取 settings.json(第 4 次)
# 短信发给 小红:你的验证码是 4821
# 从 postgres://report-db/courses 生成日报
# 共读取配置 4 次

TypeScript 版结构一样,config.ts 里的写法对应 const SETTINGS = new Config():

let readCount = 0;
function readConfigFile(): Record<string, string> { // 真实项目里是 JSON.parse(readFileSync("settings.json", "utf8"))
readCount += 1;
console.log(`读取 settings.json(第 ${readCount} 次)`);
return { currency: "CNY", sms_max_length: "30", report_db: "postgres://report-db/courses" };
}
class Config {
readonly values: Record<string, string>;
constructor() {
this.values = readConfigFile();
}
get(key: string): string {
return this.values[key];
}
}
// config.ts 里的另一种写法:导入时就读好,放进模块级全局变量
const SETTINGS = new Config();
// orders.ts
function placeOrder(user: string, price: number): void {
const currency = new Config().get("currency");
console.log(`${user} 下单:${price} ${currency}`);
}
// notifications.ts
function sendSms(user: string, text: string): void {
const limit = Number(new Config().get("sms_max_length"));
if (text.length > limit) {
console.log(`短信超过 ${limit} 字,未发送`);
return;
}
console.log(`短信发给 ${user}:${text}`);
}
// reports.ts
function dailyReport(): void {
console.log(`从 ${SETTINGS.get("report_db")} 生成日报`);
}
placeOrder("小明", 199);
placeOrder("小红", 299);
sendSms("小红", "你的验证码是 4821");
dailyReport();
console.log(`共读取配置 ${readCount} 次`);
// 输出:
// 读取 settings.json(第 1 次)
// 读取 settings.json(第 2 次)
// 小明 下单:199 CNY
// 读取 settings.json(第 3 次)
// 小红 下单:299 CNY
// 读取 settings.json(第 4 次)
// 短信发给 小红:你的验证码是 4821
// 从 postgres://report-db/courses 生成日报
// 共读取配置 4 次

本地读一个小 JSON 文件,多读几遍用户感觉不到。每个函数要什么自己拿,一目了然;SETTINGS 导入就能用。大多数项目的第一版就是这两种之一,而且一直用得好好的。

公司把配置统一搬到了配置中心(configuration center):配置不再从本地文件读,每读一次都是一次网络请求,要等,而且配置中心按应用限制调用次数,超出的请求会被拒绝;同时,测试里要能换成一份假配置。比如把短信上限设成 10 字来测“超长不发”,测试不能真的去请求配置中心。

要改的地方有五处:

  1. read_config_file 换成 fetch_remote_config。这一处任何写法都躲不掉。
  2. place_order 里的 Config()。现在每下一单就请求一次配置中心,订单越多调用越多,下单高峰时最先撞上限额。要让它共享一个实例,可 Config 自己不会“共享”,只能在 orders.py 里自己缓存一份:
orders.py
_config: Config | None = None
def place_order(user: str, price: int) -> None:
global _config
if _config is None: # 新增:自己缓存一份
_config = Config()
currency = _config.get("currency")
...
  1. send_sms:把这段缓存原样抄到 notifications.py。抄完以后,请求次数从“每次调用一次”变成“每个模块一次”,还是不止一次。第四个模块的作者不知道这个约定,又回到每次都请求。
  2. SETTINGS = Config()。它只请求一次,毛病出在“导入时”。任何代码只要 import reports,哪怕只想用里面一个小函数,都会先发一次网络请求。CI(持续集成,continuous integration)机器连不到配置中心,测试文件在导入这一步就报错,一个测试都没跑。导入顺序也开始要紧:配置中心的地址和令牌从环境变量读,main.py 要先 load_env();有人把 import reports 挪到了 main.py 最上面,请求就在 load_env() 之前发了出去,用的是空地址。
  3. 测试里换假配置。send_sms 在函数里面自己 Config(),测试没法把假配置递进去,只能用 mock 替换 notifications 模块里的 Config 这个名字,测试得知道业务函数内部用了哪个类。SETTINGS 更麻烦:from config import SETTINGS 把对象绑到了 reports 自己的名字上,替换 config.SETTINGS 对 reports 不起作用,得替换 reports.SETTINGS,导入它的模块有几个,就要换几个。

TS 版五处一样,第 4、5 处还更难办。ES 模块的 import 语句会先于模块里的其他代码执行,loadEnv() 写在 import 前面也没用;导入进来的名字是只读的,测试里不能直接赋值替换,只能靠测试框架的模块替换功能。

问题的形状是:配置什么时候读、读几次、测试时换成哪一份,散在每个调用点和“导入”那一刻,没有一个地方说了算。

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

图中 ①–⑤ 是上面列的五处;place_order、send_sms 各建一份 Config,daily_report 用的是 SETTINGS。

单例的做法是让 Config 自己管“只有一个”这件事。唯一的实例存在类字段 _instance 上;业务代码改调类方法 Config.instance():第一次调用时请求配置中心、创建实例、存起来,之后直接返回它。用到时才创建,叫惰性创建(lazy initialization)。再加一个 set_instance,让测试能换掉这个实例:

from __future__ import annotations
fetch_count = 0
def fetch_remote_config() -> dict[str, str]: # 真实项目里是一次 HTTP 请求:慢,还有调用次数限制
global fetch_count
fetch_count += 1
print(f"请求配置中心(第 {fetch_count} 次)")
return {"currency": "CNY", "sms_max_length": "30", "report_db": "postgres://report-db/courses"}
class Config:
_instance: Config | None = None # 唯一的实例存在类上
def __init__(self, values: dict[str, str]) -> None: # 约定:业务代码不直接调用,用 instance()
self._values = values
@classmethod
def instance(cls) -> Config:
if cls._instance is None: # 第一次调用才请求配置中心(惰性创建)
cls._instance = cls(fetch_remote_config())
return cls._instance
@classmethod
def set_instance(cls, config: Config | None) -> None: # 测试用:换成假配置,传 None 还原
cls._instance = config
@classmethod
def from_values(cls, values: dict[str, str]) -> Config: # 不联网,用一份现成的值造配置
return cls(values)
def get(self, key: str) -> str:
return self._values[key]
# orders.py
def place_order(user: str, price: int) -> None:
currency = Config.instance().get("currency")
print(f"{user} 下单:{price} {currency}")
# notifications.py
def send_sms(user: str, text: str) -> None:
limit = int(Config.instance().get("sms_max_length"))
if len(text) > limit:
print(f"短信超过 {limit} 字,未发送")
return
print(f"短信发给 {user}:{text}")
# reports.py
def daily_report() -> None:
print(f"从 {Config.instance().get('report_db')} 生成日报")
if __name__ == "__main__":
print("程序启动,还没有请求配置中心")
place_order("小明", 199)
place_order("小红", 299)
send_sms("小红", "你的验证码是 4821")
daily_report()
print(f"共请求配置中心 {fetch_count} 次")
# 测试里:换成假配置,不联网
Config.set_instance(Config.from_values({"currency": "CNY", "sms_max_length": "10", "report_db": "test"}))
send_sms("小红", "你的验证码是 4821")
Config.set_instance(None) # 测试结束要还原
# 输出:
# 程序启动,还没有请求配置中心
# 请求配置中心(第 1 次)
# 小明 下单:199 CNY
# 小红 下单:299 CNY
# 短信发给 小红:你的验证码是 4821
# 从 postgres://report-db/courses 生成日报
# 共请求配置中心 1 次
# 短信超过 10 字,未发送

TS 版用 private constructor 把构造函数关起来。为了和 Python 版对得上,这里的请求写成同步的;真实的网络请求是异步的,6.5 多线程那一段会说到它的影响:

let fetchCount = 0;
function fetchRemoteConfig(): Record<string, string> { // 真实项目里是一次 HTTP 请求:慢,还有调用次数限制
fetchCount += 1;
console.log(`请求配置中心(第 ${fetchCount} 次)`);
return { currency: "CNY", sms_max_length: "30", report_db: "postgres://report-db/courses" };
}
class Config {
private static _instance: Config | null = null; // 唯一的实例存在类上
private constructor(private readonly values: Record<string, string>) {} // 类外面 new Config(...) 编译不过
static instance(): Config {
if (Config._instance === null) { // 第一次调用才请求配置中心(惰性创建)
Config._instance = new Config(fetchRemoteConfig());
}
return Config._instance;
}
static setInstance(config: Config | null): void { // 测试用:换成假配置,传 null 还原
Config._instance = config;
}
static fromValues(values: Record<string, string>): Config { // 不联网,用一份现成的值造配置
return new Config(values);
}
get(key: string): string {
return this.values[key];
}
}
// orders.ts
function placeOrder(user: string, price: number): void {
const currency = Config.instance().get("currency");
console.log(`${user} 下单:${price} ${currency}`);
}
// notifications.ts
function sendSms(user: string, text: string): void {
const limit = Number(Config.instance().get("sms_max_length"));
if (text.length > limit) {
console.log(`短信超过 ${limit} 字,未发送`);
return;
}
console.log(`短信发给 ${user}:${text}`);
}
// reports.ts
function dailyReport(): void {
console.log(`从 ${Config.instance().get("report_db")} 生成日报`);
}
console.log("程序启动,还没有请求配置中心");
placeOrder("小明", 199);
placeOrder("小红", 299);
sendSms("小红", "你的验证码是 4821");
dailyReport();
console.log(`共请求配置中心 ${fetchCount} 次`);
// 测试里:换成假配置,不联网
Config.setInstance(Config.fromValues({ currency: "CNY", sms_max_length: "10", report_db: "test" }));
sendSms("小红", "你的验证码是 4821");
Config.setInstance(null); // 测试结束要还原
// 输出:
// 程序启动,还没有请求配置中心
// 请求配置中心(第 1 次)
// 小明 下单:199 CNY
// 小红 下单:299 CNY
// 短信发给 小红:你的验证码是 4821
// 从 postgres://report-db/courses 生成日报
// 共请求配置中心 1 次
// 短信超过 10 字,未发送

单例版:Config 与它的调用点

每个调用点都向同一个类要实例。

单例的调用顺序

看顺序图:第二次 instance() 不再请求,set_instance(fake) 后拿到假配置。

逐条对照 6.3 的五处:

  • 第 1 处,fetch_remote_config:只在 instance() 里被调用,整个程序只有这一处。
  • 第 2、3 处,各模块的缓存:不用抄了。每个调用点从 Config() 改成 Config.instance(),这一遍躲不掉,但改完任何模块都不用写缓存,第四个模块直接写 Config.instance() 就对。运行结果里配置中心只请求了 1 次。
  • 第 4 处,导入时联网:导入时什么都不会发生,请求发生在第一次下单时。导入顺序也不再要紧,只要第一次调用 Config.instance() 在 load_env() 之后。想在启动时就发现连不上配置中心,可在 main() 里 load_env() 之后主动调一次,创建时机由你定。
  • 第 5 处,测试换假配置:一行 Config.set_instance(...),所有模块拿到的都是假的,测试不用知道是哪个函数在读配置。运行结果最后一行就是效果:上限 10 字,11 个字的短信被拦下。

关于私有构造:Python 没有私有构造函数,__init__ 上的注释只是约定,谁写一句 Config(fetch_remote_config()) 照样能跑。TS 的 private constructor 能拦住类外面的 new Config(...),可类里的 fromValues 照样能造出第二个 Config。它不联网,单例要守住的是“请求配置中心”只发生一次。

也要说清它没做到什么。哪种写法都得有一处负责创建,SETTINGS = Config() 也是。单例换来的是两样:用到时才创建,和一个所有模块都知道去哪儿拿的入口。代价是签名里看不出它读了配置,测试换了实例还得换回来,6.8 细说。

GoF 对单例的定义是:保证一个类仅有一个实例,并提供一个访问它的全局访问点。

换成本章的话说:“仅有一个”靠的是 _instance 字段加 instance() 里那个 if;“全局访问点”就是 Config.instance(),任何模块不用传参就能拿到。这是两件事:本章的需求要的是前一半;后一半是顺带来的,6.8 的麻烦大多出自它。

单例的结构:Singleton 自己持有唯一实例,getInstance 总是返回它

图里是 GoF 的 Singleton;set_instance 和 from_values 是为测试加的,不在 GoF 原始结构里。

角色 在本章案例里是谁 职责
单例(Singleton) Config 用类字段保存唯一的实例;instance() 第一次调用时创建它,之后总是返回同一个

常见场景:

  • 配置:本章的例子,读一次,到处用。
  • 日志:同一个名字的 logger 全进程只有一个,6.7 例一。
  • 连接池:数据库连接建起来慢、数量有上限,整个进程共用一个池,6.10 的练习就是它。
  • 进程内缓存:所有模块共用一份才有意义,各建各的,谁也查不到别人存进去的数据。

多线程下要加锁。Python 的 Web 服务常用多线程处理请求。两个线程同时第一次调用 instance(),都看到 _instance 是 None,于是都去请求配置中心:请求发了两次,后建好的那个覆盖先建好的,先拿到实例的线程手里是另一个对象。加一把锁:

import threading
class Config:
_instance: "Config | None" = None
_lock = threading.Lock()
@classmethod
def instance(cls) -> "Config":
if cls._instance is None: # 建好之后不用每次都抢锁
with cls._lock:
if cls._instance is None: # 拿到锁后再查一次:等锁期间别的线程可能已经建好了
cls._instance = cls(fetch_remote_config())
return cls._instance

这种查两次的写法叫双重检查锁定(double-checked locking)。TS 没有这个问题:JavaScript 代码在一个线程上执行,同步的 instance() 不会被别的代码打断。但真实的请求是异步的,instance() 一旦变成 async,两个调用方都可能在第一个请求返回之前看到 _instance 为空,照样请求两次。解决办法是把第一次请求的 Promise 存起来,后来的调用方等同一个 Promise。

网上还常见重写 __new__ 让 Config() 每次返回同一个对象的写法,坑是 __init__ 每次照样被调用,构造里读配置就每次都读。

Python 的模块本身就是单例。模块第一次被导入时执行顶层代码,模块对象存进 sys.modules;之后再导入,直接拿缓存里的那个。所以 config.py 里写一行 settings = fetch_remote_config(),所有 from config import settings 拿到的都是同一个对象。要惰性,就放进一个加了 functools.lru_cache 的函数:

from functools import lru_cache
fetch_count = 0
def fetch_remote_config() -> dict[str, str]:
global fetch_count
fetch_count += 1
print(f"请求配置中心(第 {fetch_count} 次)")
return {"currency": "CNY", "sms_max_length": "30"}
# 写法一:模块就是单例。真实项目里是两个文件:
# config.py settings = fetch_remote_config()
# orders.py from config import settings
# 下面用一个函数模拟 import 的"只执行一次"
_modules: dict[str, dict[str, str]] = {}
def import_config() -> dict[str, str]:
if "config" not in _modules: # 相当于 sys.modules 里还没有 config
_modules["config"] = fetch_remote_config() # 相当于执行 config.py 的顶层代码
return _modules["config"]
# 写法二:惰性版本,第一次调用才请求,之后返回缓存的结果
@lru_cache(maxsize=None)
def get_settings() -> dict[str, str]:
return fetch_remote_config()
if __name__ == "__main__":
print(f"orders 导入:{import_config()['currency']}")
print(f"notifications 导入:{import_config()['sms_max_length']}")
print(f"写法二还没被调用,共请求 {fetch_count} 次")
print(f"第一次调用:{get_settings()['currency']}")
print(f"两次拿到同一个对象:{'是' if get_settings() is get_settings() else '否'}")
print(f"共请求 {fetch_count} 次")
# 输出:
# 请求配置中心(第 1 次)
# orders 导入:CNY
# notifications 导入:30
# 写法二还没被调用,共请求 1 次
# 请求配置中心(第 2 次)
# 第一次调用:CNY
# 两次拿到同一个对象:是
# 共请求 2 次

TS 一样:ES 模块不管被导入多少次,顶层代码只执行一次,config.ts 里 export const config = fetchRemoteConfig() 就是一个模块单例。惰性版本没有现成的 lru_cache,写一个几行的 once:

let fetchCount = 0;
function fetchRemoteConfig(): Record<string, string> {
fetchCount += 1;
console.log(`请求配置中心(第 ${fetchCount} 次)`);
return { currency: "CNY", sms_max_length: "30" };
}
// 写法一:模块就是单例。真实项目里是两个文件:
// config.ts export const config = fetchRemoteConfig();
// orders.ts import { config } from "./config";
// 下面用一个函数模拟 import 的"只执行一次"
const modules = new Map<string, Record<string, string>>();
function importConfig(): Record<string, string> {
let config = modules.get("config"); // 相当于查已经加载过的模块
if (config === undefined) {
config = fetchRemoteConfig(); // 相当于执行 config.ts 的顶层代码
modules.set("config", config);
}
return config;
}
// 写法二:惰性版本,第一次调用才请求,之后返回同一个结果
function once<T>(fn: () => T): () => T {
let cache: { value: T } | null = null;
return () => (cache ??= { value: fn() }).value;
}
const getSettings = once(fetchRemoteConfig);
console.log(`orders 导入:${importConfig().currency}`);
console.log(`notifications 导入:${importConfig().sms_max_length}`);
console.log(`写法二还没被调用,共请求 ${fetchCount} 次`);
console.log(`第一次调用:${getSettings().currency}`);
console.log(`两次拿到同一个对象:${getSettings() === getSettings() ? "是" : "否"}`);
console.log(`共请求 ${fetchCount} 次`);
// 输出:
// 请求配置中心(第 1 次)
// orders 导入:CNY
// notifications 导入:30
// 写法二还没被调用,共请求 1 次
// 请求配置中心(第 2 次)
// 第一次调用:CNY
// 两次拿到同一个对象:是
// 共请求 2 次

结构还是同一个:一个实例,一个拿它的地方。语言省掉的是 _instance 字段和 instance() 里的那个 if,“只执行一次”由 import 机制或 lru_cache 替你记着,连 Config 这个类都可以不要。

但省掉的不只是代码。写法一回到了 6.3 第 4 处的“导入时执行”,那几个毛病原样都在。写法二没有这个问题,可也没有 set_instance:测试想换假配置,只能先 get_settings.cache_clear(),再用 mock 替换 fetch_remote_config。所以判断标准是:创建便宜、导入时执行也无妨、测试不需要换掉它时,一个模块级变量就够了,比如只读几个环境变量的配置。要联网、测试里要换的,用 6.4 的写法,或者 6.8 末尾的传参。第 29 章还会讨论模块替掉单例这件事。

例一:logging.getLogger 的注册表(Python)

Section titled “例一:logging.getLogger 的注册表(Python)”

Python 标准库的 logging 里,logging.getLogger("app.orders") 不管在哪个模块、调用多少次,拿到的都是同一个 Logger 对象,官方文档写明了这一点。所以惯用写法是每个模块顶部写 logger = logging.getLogger(__name__),启动脚本里按名字调级别,各模块手里的 logger 立刻跟着变。它不是“一个类只有一个实例”,而是“每个名字只有一个实例”:标准库里有一个全进程共用的管理器 Logger.manager,它的 loggerDict 是名字到 logger 的字典。这种按键各一个的变体常被叫作多例(Multiton)或注册表(Registry)。简化版(方法名照抄标准库,默认级别简化成 INFO):

DEBUG, INFO, WARNING = 10, 20, 30
class Logger:
def __init__(self, name: str) -> None:
self.name = name
self.level = INFO
def setLevel(self, level: int) -> None:
self.level = level
def info(self, msg: str) -> None:
if self.level <= INFO:
print(f"INFO:{self.name}:{msg}")
def warning(self, msg: str) -> None:
if self.level <= WARNING:
print(f"WARNING:{self.name}:{msg}")
class Manager: # 注册表:名字到 Logger 的字典
def __init__(self) -> None:
self.loggerDict: dict[str, Logger] = {}
def getLogger(self, name: str) -> Logger:
if name not in self.loggerDict: # 第一次要这个名字才创建
self.loggerDict[name] = Logger(name)
return self.loggerDict[name]
manager = Manager() # 整个进程只有这一个注册表
def getLogger(name: str) -> Logger:
return manager.getLogger(name)
# orders.py、payments.py 各自在模块顶部要一个 logger
orders_log = getLogger("app.orders")
payments_log = getLogger("app.payments")
if __name__ == "__main__":
orders_log.info("小明 下单 199 元")
getLogger("app.orders").setLevel(WARNING) # 启动脚本按名字调级别,手里没有 orders_log
orders_log.info("小红 下单 299 元") # 被过滤了:两处拿到的是同一个对象
orders_log.warning("库存不足")
payments_log.info("支付成功")
print(f"注册表里的名字:{list(manager.loggerDict)}")
# 输出:
# INFO:app.orders:小明 下单 199 元
# WARNING:app.orders:库存不足
# INFO:app.payments:支付成功
# 注册表里的名字:['app.orders', 'app.payments']

getLogger(name) 就是带了参数的 Config.instance(),loggerDict 对应 _instance,只是从“一个”变成了“每个名字一个”。真实实现还要加锁,还要处理 app 和 app.orders 之间的父子关系。

例二:Redux 和 Pinia 的应用级 store(TypeScript)

Section titled “例二:Redux 和 Pinia 的应用级 store(TypeScript)”

前端状态管理库把整个应用共享的状态放在一个 store 里。Redux 的 createStore(现在官方推荐 Redux Toolkit 的 configureStore)本身不是单例,调几次就造几个;“整个应用一个”是应用自己的写法:在 store.ts 里创建一次并导出,各组件导入的都是这一个。Pinia 类似:createPinia() 造一个实例,app.use(pinia) 装到应用上,各组件的 store 都挂在它下面。这正是 6.6 的模块单例。简化版:

type Listener = () => void;
interface Store<S, A> {
getState(): S;
dispatch(action: A): void;
subscribe(listener: Listener): void;
}
function createStore<S, A>(reducer: (state: S, action: A) => S, initial: S): Store<S, A> {
let state = initial;
const listeners: Listener[] = [];
return {
getState: () => state,
dispatch(action) {
state = reducer(state, action);
listeners.forEach((listener) => listener());
},
subscribe(listener) {
listeners.push(listener);
},
};
}
interface CartState { readonly items: readonly string[] }
type CartAction = { type: "cart/add"; course: string };
function cartReducer(state: CartState, action: CartAction): CartState {
return { items: [...state.items, action.course] };
}
// store.ts:整个应用只在这里创建一次(真实项目里是 export const store = configureStore({ ... }))
const store = createStore(cartReducer, { items: [] });
// CartBadge 组件:订阅变化,刷新购物车角标
store.subscribe(() => console.log(`角标:${store.getState().items.length}`));
// CoursePage 组件:点"加入购物车"
function onAddClick(course: string): void {
store.dispatch({ type: "cart/add", course });
}
onAddClick("Python 入门");
onAddClick("设计模式");
console.log(`结算页看到:${store.getState().items.join("、")}`);
// 输出:
// 角标:1
// 角标:2
// 结算页看到:Python 入门、设计模式

const store = createStore(...) 那一行对应 6.6 的 export const config = ...,两个组件互不传参,靠同一个模块拿到同一个 store。要注意服务端渲染(server-side rendering):服务器上一个进程同时处理许多用户的请求,Redux 和 Pinia 的文档都提醒过,这时 store 要每个请求新建一个,做成模块级单例的话,一个用户的购物车可能出现在另一个用户的页面上。这正是 6.8 要说的事。

单例大概是最常被滥用的模式:Config.instance() 在任何地方都能写,比把参数一层层传下去省事得多。下面四种代价,都付在“能直接拿”这一半上。

  1. 函数签名藏起了依赖。看这个函数:
def calc_refund(order: Order) -> int:
days = int(Config.instance().get("refund_days"))
...

签名看起来是纯计算:给一个订单,算出能退多少钱。财务同事写对账脚本,在自己电脑上 from orders import calc_refund 直接调用,脚本卡了半天,最后报连接配置中心超时。他得读完 calc_refund 和它调用的每个函数,才知道这个“算钱”的函数要联网。代码审查时,看调用处 calc_refund(order) 也看不出它读了配置。

  1. 测试单独跑通过,一起跑失败。测试 A 用 set_instance 换上一份短信上限 10 字的假配置,测完忘了还原。测试 B 检查“28 字的短信能发出去”,用的是 conftest.py 里统一设好的测试配置(上限 30 字)。单独跑 B,通过;整套一起跑,A 恰好排在 B 前面,B 拿到的是 A 留下的那份,失败。B 的代码里没有 A 的影子,看失败信息完全想不到是 A 干的;有人给 A 改了个文件名,执行顺序一变,失败又“自己好了”。同一个进程里的测试共用这个实例,谁动了它,都要在 finally 或测试框架的清理步骤里换回来,漏一个就埋一个雷。

  2. 全局可变状态。本章的 Config 只读,还算安全。一旦有人给它加了 set:海外活动页面的处理函数里写了一句 Config.instance().set("currency", "USD"),本意只想影响这一个页面,可所有模块拿的都是同一个对象,从这一刻起,这个进程里的每一单都按美元结算,直到重启。更糟的是把“当前登录用户”这类每个请求各不相同的数据放进单例:多线程的服务器上,请求 A 刚把当前用户设成小明,还没处理完,请求 B 把它改成了小红,A 返回的页面里显示的是小红的订单。每个请求自己的数据应该跟着请求走。

  3. 多进程部署时,“唯一”只在一个进程里成立。Python 的 Web 服务常用 gunicorn 起好几个工作进程,Node.js 服务也常开多个进程或多个容器。每个进程都有自己的一份 _instance,彼此看不见。假设开了 4 个进程:运营在配置中心把短信上限从 30 改成 40,后台有一个“刷新配置”的接口,里面调 Config.set_instance(None) 让下次重新拉取;这个请求只会落到 4 个进程中的一个,于是一部分请求按 40 字算,其余的还按 30 字算,同一个用户刷新两次,可能看到不同的结果。再比如用单例在内存里数“每个手机号每天最多发 5 条验证码”,4 个进程各数各的,一个手机号最多能发 20 条。真正需要全局唯一的东西,要放进所有进程共享的地方,比如数据库或 Redis。

替代办法是把实例当参数传进去:

def send_sms(user: str, text: str, config: Config) -> None:
limit = int(config.get("sms_max_length"))
...
def main() -> None:
load_env()
config = Config.from_values(fetch_remote_config()) # 只在入口创建一次
send_sms("小红", "你的验证码是 4821", config)

配置仍然只请求一次,“只有一个”由程序入口保证,而不是由类自己保证。签名里写明了 send_sms 要用配置;测试直接传一个假配置进去,不碰全局状态,也就不用还原。代价是要一层层往下传,层数深了,中间每一层都得多一个参数。这就是依赖注入(Dependency Injection)的雏形,FastAPI 的 Depends 这类机制可以替你做传递的活,第 30 章再讲。

日志、连接池这类确实全进程一份的东西,用单例是合理的。即使这样,也尽量只在入口附近调一次 instance(),往下靠参数传。

要分清的是“只有一份对象”的四种拿法:

写法 例子 谁创建 测试时能否替换 何时创建
单例(Singleton) Config.instance() 类自己,在 instance() 里 能,set_instance 换掉,用完要还原 第一次调用 instance() 时
全局变量 CONFIG = None,启动代码里再赋值,各模块用 config.CONFIG 访问 谁都可以赋值 能,谁都能改,所以谁都可能改坏 赋值那行代码执行时
模块级实例 settings = fetch_remote_config()、export const config = fetchRemoteConfig() 模块的顶层代码 难:Python 要在每个导入它的模块里替换名字,TS 要靠测试框架替换整个模块 第一次导入时
依赖注入(Dependency Injection) send_sms(user, text, config) 程序入口或框架 能,调用时传另一个,不动全局状态 入口决定

一句话区分:前三种都是“只有一份,到处都能直接拿”,差别在谁创建、什么时候创建;依赖注入也可以只有一份,但不让人到处直接拿,用的人必须从参数里接。

后台有一个数据库连接池 ConnectionPool,ConnectionPool.open(5) 会建立 5 个数据库连接,比较慢。现在下单函数 create_order(user, course) 每次调用都自己 open(5) 一次,用完就扔。请把它改成可替换的单例:

  • ConnectionPool 加类方法 instance()(第一次调用才 open(5))和 set_instance(pool);execute(sql, params) 返回一行“执行成功:…”。
  • create_order 改用 ConnectionPool.instance(),执行一条带两个占位符的 INSERT。
  • 写一个测试函数 test_create_order_uses_pool():用继承 ConnectionPool 的 FakePool 只记录 SQL、不建连接;set_instance 换上它,调用 create_order,检查记下的 SQL,最后一定还原。
  • 入口先跑这个测试,再正常下一单,确认“建立 5 个数据库连接”只出现一次。Python 和 TS 各写一版。

提示:还原写在 finally 里。TS 里构造函数如果写成 private,FakePool 就继承不了,编译报错;写成 protected,类外面不能 new,子类可以继承。做完想一想:如果改成把连接池作为参数传给 create_order,这个测试要怎么写,还需要 finally 吗?

  • 单例让类自己保存唯一的实例,通过 instance() 拿:第一次调用才创建,之后总是同一个。
  • 它换来的是惰性创建和一个统一的入口;任何写法都要有一个地方负责创建,单例只是把这个地方放在了类自己身上。
  • 给它加一个 set_instance,测试才换得掉;换了就要还原,否则状态会漏到下一个测试。
  • 模块本身只执行一次,模块级变量是最简洁的单例,代价是导入时就执行,测试时难替换。
  • 单例的麻烦都出在“到处都能直接拿”:依赖藏进函数内部、全局可变状态、多进程下并不唯一。能传参数,就传参数。