跳转到内容

将查询函数和修改函数分离

将查询函数和修改函数分离(Separate Query from Modifier)把一个“既返回结果、又改了状态”的函数拆成两个:一个只回答问题,一个只做修改。 它解决的痛是:调用方只想知道一个答案,却不得不连带触发一次修改,于是谁都不敢随便调用它。

小郑在给一栋写字楼做安保系统。闸机每刷过一批人,系统就拿这批人的名字去和黑名单比对:发现黑名单上的人,就拉响警报,并在值班台的屏幕上显示“可疑人员:某某”。黑名单上目前有两个人,老刀和阿强。小郑写了一个函数 alert_for_miscreant,一边找人一边报警,顺手把找到的名字返回给值班台用。

"""安保系统:重构前。"""
alarms: list[str] = []
def set_off_alarms(name: str) -> None:
alarms.append(f"警报:发现 {name}")
def alert_for_miscreant(people: list[str]) -> str:
for p in people:
if p == "老刀":
set_off_alarms(p)
return "老刀"
if p == "阿强":
set_off_alarms(p)
return "阿强"
return ""
def check_security(people: list[str]) -> str:
found = alert_for_miscreant(people)
return f"可疑人员:{found}" if found else "一切正常"
if __name__ == "__main__":
print(check_security(["小王", "阿强", "老刀"]))
print(check_security(["小王", "小李"]))
for alarm in alarms:
print(alarm)
# 输出:
# 可疑人员:阿强
# 一切正常
# 警报:发现 阿强
// 安保系统:重构前。
export const alarms: string[] = [];
function setOffAlarms(name: string): void {
alarms.push(`警报:发现 ${name}`);
}
export function alertForMiscreant(people: string[]): string {
for (const p of people) {
if (p === "老刀") {
setOffAlarms(p);
return "老刀";
}
if (p === "阿强") {
setOffAlarms(p);
return "阿强";
}
}
return "";
}
export function checkSecurity(people: string[]): string {
const found = alertForMiscreant(people);
return found ? `可疑人员:${found}` : "一切正常";
}
console.log(checkSecurity(["小王", "阿强", "老刀"]));
console.log(checkSecurity(["小王", "小李"]));
for (const alarm of alarms) {
console.log(alarm);
}
// 输出:
// 可疑人员:阿强
// 一切正常
// 警报:发现 阿强

测试通过值班台用的入口 check_security 来测,同时检查警报到底响了几次。这个入口的签名在重构前后不变,所以同一份测试能跑两个版本:

import unittest
from target import alarms, check_security
class SecurityTest(unittest.TestCase):
def setUp(self):
alarms.clear()
def test_reports_first_miscreant(self):
self.assertEqual(check_security(["小王", "阿强", "老刀"]), "可疑人员:阿强")
def test_alarm_goes_off_once(self):
check_security(["小王", "阿强", "老刀"])
self.assertEqual(alarms, ["警报:发现 阿强"])
def test_no_miscreant_no_alarm(self):
self.assertEqual(check_security(["小王", "小李"]), "一切正常")
self.assertEqual(alarms, [])
if __name__ == "__main__":
unittest.main()
import test, { beforeEach } from "node:test";
import assert from "node:assert/strict";
import { alarms, checkSecurity } from "./target";
beforeEach(() => {
alarms.length = 0;
});
test("报出第一个可疑人员", () => {
assert.equal(checkSecurity(["小王", "阿强", "老刀"]), "可疑人员:阿强");
});
test("警报只响一次", () => {
checkSecurity(["小王", "阿强", "老刀"]);
assert.deepEqual(alarms, ["警报:发现 阿强"]);
});
test("没有可疑人员就不报警", () => {
assert.equal(checkSecurity(["小王", "小李"]), "一切正常");
assert.deepEqual(alarms, []);
});

如果现在要加一个需求:大堂门口的电子屏也要显示“今日访客中有黑名单人员:某某”,但它只是个提示屏,不该拉警报。警报已经在闸机那里响过一次了,大堂屏每刷新一次就再响一次,保安会疯掉。

小郑翻遍代码,能告诉他“这批人里有没有黑名单上的人、是谁”的只有 alert_for_miscreant,而调用它就会报警。于是他只剩两条路:把那段循环复制一份,删掉报警的两行,叫 find_name_for_screen;或者给 alert_for_miscreant 加一个参数 silent=True。前者让黑名单比对的逻辑有了两份,下个月黑名单加第三个人时,很可能只改了一份,大堂屏和闸机给出不一样的结论。后者是第 44 章要讲的标记参数,函数从此一半时间报警、一半时间不报警,调用处的 True 和 False 谁也看不懂。

第二个后果更隐蔽。看到一个返回字符串的函数,读者的默认假设是“调用它是安全的,多调一次只是多算一次”。有人在日志模块里写下这样一行:

if alert_for_miscreant(people):
log(f"发现 {alert_for_miscreant(people)}")

这行看起来只是偷懒没存变量,实际上让警报响了两次。代码评审很难抓到它,因为函数名里的 alert 并没有大声宣布“我会改动世界”,返回值又让它看上去像一个查询。测试也很难抓到,除非测试刚好去数警报次数。

第三个问题在测试本身。要测“名单里有没有阿强”这条规则,就必须先准备好 alarms 列表、测完再清空,因为每一次比对都会往里面写东西。一个本来是纯逻辑的判断,测起来却要照顾副作用。

用第 2 章的名字说,这是可变数据:一个看上去只读的调用悄悄改了共享状态 alarms,读代码的人得追进函数体才知道调用它的代价。而那份被迫复制出来的比对循环,会再添一处重复代码。

"""安保系统:重构后。"""
MISCREANTS = ("老刀", "阿强")
alarms: list[str] = []
def set_off_alarms(name: str) -> None:
alarms.append(f"警报:发现 {name}")
def find_miscreant(people: list[str]) -> str:
for p in people:
if p in MISCREANTS:
return p
return ""
def alert_for_miscreant(people: list[str]) -> None:
found = find_miscreant(people)
if found:
set_off_alarms(found)
def check_security(people: list[str]) -> str:
found = find_miscreant(people)
alert_for_miscreant(people)
return f"可疑人员:{found}" if found else "一切正常"
if __name__ == "__main__":
print(check_security(["小王", "阿强", "老刀"]))
print(check_security(["小王", "小李"]))
for alarm in alarms:
print(alarm)
# 输出:
# 可疑人员:阿强
# 一切正常
# 警报:发现 阿强
// 安保系统:重构后。
const MISCREANTS = ["老刀", "阿强"];
export const alarms: string[] = [];
function setOffAlarms(name: string): void {
alarms.push(`警报:发现 ${name}`);
}
export function findMiscreant(people: string[]): string {
for (const p of people) {
if (MISCREANTS.includes(p)) return p;
}
return "";
}
export function alertForMiscreant(people: string[]): void {
const found = findMiscreant(people);
if (found) setOffAlarms(found);
}
export function checkSecurity(people: string[]): string {
const found = findMiscreant(people);
alertForMiscreant(people);
return found ? `可疑人员:${found}` : "一切正常";
}
console.log(checkSecurity(["小王", "阿强", "老刀"]));
console.log(checkSecurity(["小王", "小李"]));
for (const alarm of alarms) {
console.log(alarm);
}
// 输出:
// 可疑人员:阿强
// 一切正常
// 警报:发现 阿强

输出没变,三个测试照样通过。

重构前值班台只能通过会报警的 alert_for_miscreant 拿到名字;重构后名字来自不报警的 find_miscreant,大堂屏这个新调用方只连到它

大堂屏的需求现在是两行,不碰警报:

def lobby_screen(people: list[str]) -> str:
found = find_miscreant(people)
return f"今日访客中有黑名单人员:{found}" if found else ""

黑名单比对只有一份,在 find_miscreant 里。顺带地,原来两个几乎一样的 if 合成了一个对 MISCREANTS 的查找,加第三个人只改这一行常量。

日志模块里那种“调两次”的写法不再危险:find_miscreant 调多少次都只是多算几次。alert_for_miscreant 不再返回任何东西,想从它拿名字的人会在类型检查时就碰壁,返回值是 None / void,自然会去找那个查询函数。

测“名单里有没有阿强”这条规则,现在可以直接测 find_miscreant,不需要准备和清空 alarms。

你可能注意到 check_security 里比对做了两遍:自己调一次 find_miscreant,alert_for_miscreant 里又调一次。名单只有几个人、一批访客几十人,这点重复的开销测不出来。真要在意,可以把 check_security 改成拿到 found 之后直接调 set_off_alarms(found)。但先把查询和修改分开,再看性能,顺序不要反过来。

每一步之后都跑测试。

  1. 复制原函数,按查询的意思起名。alert_for_miscreant 的返回值是“找到的名字”,所以新函数叫 find_miscreant。名字想不好时,看调用方拿这个返回值做什么:值班台用它显示名字,那它就是“找人”。
  2. 从新函数里删掉所有副作用:两处 set_off_alarms(p)。现在它是一个纯查询,可以跑一下测试,确认什么都没坏,因为还没有人调用它。
  3. 逐个找到原函数的调用方,把“用返回值”的那部分改成调用查询函数,原函数的调用保留在它下一行。check_security 里的 found = alert_for_miscreant(people) 变成: found = find_miscreant(people),紧接着一行 alert_for_miscreant(people)。每改一个调用方跑一次测试。
  4. 所有调用方都不再使用原函数的返回值之后,把原函数的返回值去掉:return "老刀" 之类改成 return,返回类型改成 None / void。TS 的类型检查器会替你找出漏掉的调用方。
  5. 原函数里的比对循环和查询函数重复了,让它改为调用查询函数:found = find_miscreant(people),找到就报警。这一步消除了第 1 步复制出来的重复。

第 1 步是复制加改名,改名可以交给 IDE 的 Rename;第 4 步改返回类型要手工做,改完由类型检查或 IDE 的 Find Usages 找出还在用旧返回值的调用方。删副作用、改调用方的判断都得自己来。如果原函数有很多调用方,第 3 步可以分几次提交,中间状态完全可以发布:原函数照旧返回名字,只是调用方渐渐不用它了。

将查询函数和修改函数分离:任何有返回值的函数,都不应该有看得到的副作用;把既返回值又修改状态的函数拆成一个查询函数和一个修改函数。

“看得到的副作用”是关键词。函数内部缓存了一个计算结果、记了一条调试日志,外部观察不到差别,不算违反。往数据库写一行、发一封邮件、改了传进来的列表、拉响警报,这些调用方看得到,就算。判断方法很简单:把这个函数连调两次,结果和调一次是否一样。不一样,它就不是一个查询。

这个手法常出现在三类代码里。一是“找到就顺手处理”的循环,就像本章的例子,找东西和处理东西被写在了一起。二是“取出并标记”的业务函数,例如 next_task() 返回下一个任务并把它标成进行中,看板页想预览下一个任务就做不到。三是带懒初始化的读取函数,读的时候发现没有就创建并保存,这会让“只是看一眼”的代码在数据库里留下一条记录。

本章没有反向手法。偶尔你会想把一个查询和一个修改合在一起,那几乎总是因为并发:先查后改的两步之间,别的线程可能已经改了状态。这时候合并是为了原子性,不是为了省事,下面栈的例子会说到。

本章做的事有个名字,叫命令查询分离(Command-Query Separation,CQS),由 Bertrand Meyer 在设计 Eiffel 语言时提出:方法要么是改变状态的命令,要么是返回结果的查询,不能两者都是。它是一条设计原则,不是 GoF 的设计模式,《设计模式》书里没有对应章节。

很多语言的栈只提供 push 和 pop。pop 同时做两件事:返回栈顶、把它移除。编辑器的撤销功能用栈记录操作,菜单上要显示“撤销「输入文字」”,只是想看一眼栈顶。只有 pop 的时候,只能先弹出再压回去,忘了压回去就丢一步撤销记录。加上纯查询 peek 之后,看一眼就真的只是看一眼。下面用数组模拟一个撤销栈。

// 撤销栈:只有 pop 时,想看一眼栈顶也得先弹出再压回去。
class UndoStackV1 {
private items: string[] = [];
push(action: string): void {
this.items.push(action);
}
pop(): string | undefined {
return this.items.pop();
}
size(): number {
return this.items.length;
}
}
function undoLabelV1(stack: UndoStackV1): string {
const top = stack.pop();
if (top === undefined) return "无可撤销";
stack.push(top); // 忘了这一行,撤销记录就少一条
return `撤销「${top}」`;
}
// 加上纯查询 peek:看栈顶不再改栈。
class UndoStackV2 {
private items: string[] = [];
push(action: string): void {
this.items.push(action);
}
pop(): string | undefined {
return this.items.pop();
}
peek(): string | undefined {
return this.items[this.items.length - 1];
}
size(): number {
return this.items.length;
}
}
function undoLabelV2(stack: UndoStackV2): string {
const top = stack.peek();
return top === undefined ? "无可撤销" : `撤销「${top}」`;
}
const s1 = new UndoStackV1();
const s2 = new UndoStackV2();
for (const action of ["输入文字", "加粗"]) {
s1.push(action);
s2.push(action);
}
console.log(undoLabelV1(s1), s1.size());
console.log(undoLabelV2(s2), s2.size());
console.log(undoLabelV2(new UndoStackV2()));
// 输出:
// 撤销「加粗」 2
// 撤销「加粗」 2
// 无可撤销

注意 pop 本身并没有被拆掉。“取出并移除”是栈最核心的一个动作,拆成 peek 加一个不返回值的 remove,单线程里只是啰嗦,多线程里就是错误:两个消费者可能同时 peek 到同一个元素,然后各自 remove 一次,一个元素被处理两次,另一个被丢掉。Python 的 queue.Queue.get 也是既取又删,原因相同。命令查询分离是默认做法,原子性需要时可以破例,但破例要有这种说得出口的理由。

dict.setdefault(key, default) 是标准库里一个典型的“既查又改”:键存在就返回它的值,不存在就先把默认值存进去再返回。它的名字里有 set,用的人却常常把它当 get 用,于是一次读配置在字典里多写了一个键,这个键随后被原样序列化保存。下面对比两种写法,再看它真正适合的场合。

"""dict.setdefault:既查又改。"""
import json
def timeout_v1(config: dict) -> int:
# 只是想读超时,顺手用了 setdefault
return config.setdefault("timeout", 30)
def timeout_v2(config: dict) -> int:
# 纯查询:读不到就用默认值,字典不变
return config.get("timeout", 30)
if __name__ == "__main__":
user_config = {"host": "example.com"}
print(timeout_v1(user_config), json.dumps(user_config))
user_config = {"host": "example.com"}
print(timeout_v2(user_config), json.dumps(user_config))
# 分组时要的正是"没有就放一个空列表,再返回它"
orders = [("张三", "咖啡豆"), ("李四", "滤纸"), ("张三", "手冲壶")]
by_customer: dict[str, list[str]] = {}
for name, item in orders:
by_customer.setdefault(name, []).append(item)
print(by_customer)
# 输出:
# 30 {"host": "example.com", "timeout": 30}
# 30 {"host": "example.com"}
# {'张三': ['咖啡豆', '手冲壶'], '李四': ['滤纸']}

第一行的副作用就是本章要防的那种:保存用户配置时多出一个用户从没设过的 "timeout": 30,以后改默认值,这位用户永远停在 30。读的时候用 get,写的时候明确写,查询和修改各归各位。

分组那段则是 setdefault 该在的地方:调用方要的就是“确保有、再拿到它”,修改是目的而不是意外。即便如此,collections.defaultdict(list) 往往更清楚,它把“缺了就补”声明在字典创建的那一刻,而不是藏在每一次读取里。

返回值本身就是修改的结果时,不用拆。list.pop()、counter.increment() 返回新值、数据库插入后返回新记录的 ID,这些返回值描述的是“刚刚做了什么”,调用方需要它,而且不存在“只想知道不想改”的用法。硬拆成 insert() 加 last_inserted_id(),反而要在两次调用之间保证没有别人插入。

函数的副作用只是内部缓存或调试日志时,也不用拆,前面说过,外部看不到的副作用不算副作用。另外,如果拆开后查询函数要重复一次昂贵的计算,比如一次网络请求或一次全表扫描,先测量,再决定是让修改函数接收查询的结果做参数,还是接受重复。

手法 它关心什么 本章手法关心什么
移除标记参数(第 44 章) 一个函数靠布尔参数切换两种行为 一个函数同时回答问题和改状态;不拆的话,常见的偷懒补丁正是加一个 silent 标记
以参数取代查询(第 46 章) 函数内部读全局状态,改为从参数传入 函数内部写共享状态,把写的部分拆出去
提炼函数(第 4 章) 给一段代码起名字 拆分的依据是“有没有副作用”,第 1 步的复制其实就是一次带着复制的提炼

图书馆每天跑一次催还:超过 30 天没还书的读者发一条提醒,同时日报上要写今天催了几个人。remind_overdue 既发提醒又返回人数。把它拆成一个只数人数的查询函数 count_overdue 和一个只发提醒、不返回值的 remind_overdue,daily_report 的签名保持不变。

"""练习:图书馆催还。"""
sent: list[str] = []
def send_reminder(name: str) -> None:
sent.append(name)
def remind_overdue(members: list[dict]) -> int:
count = 0
for member in members:
if member["overdue_days"] > 30:
send_reminder(member["name"])
count += 1
return count
def daily_report(members: list[dict]) -> str:
return f"今日催还 {remind_overdue(members)} 人"
MEMBERS = [
{"name": "小林", "overdue_days": 45},
{"name": "小陈", "overdue_days": 3},
{"name": "小吴", "overdue_days": 31},
]
if __name__ == "__main__":
print(daily_report(MEMBERS))
print(sent)
# 输出:
# 今日催还 2 人
# ['小林', '小吴']
import unittest
from target import MEMBERS, daily_report, sent
class OverdueTest(unittest.TestCase):
def setUp(self):
sent.clear()
def test_report_counts_overdue(self):
self.assertEqual(daily_report(MEMBERS), "今日催还 2 人")
def test_reminders_sent_once_each(self):
daily_report(MEMBERS)
self.assertEqual(sent, ["小林", "小吴"])
def test_nobody_overdue(self):
self.assertEqual(daily_report([{"name": "小陈", "overdue_days": 3}]), "今日催还 0 人")
self.assertEqual(sent, [])
if __name__ == "__main__":
unittest.main()
// 练习:图书馆催还。
export interface Member {
name: string;
overdueDays: number;
}
export const sent: string[] = [];
function sendReminder(name: string): void {
sent.push(name);
}
export function remindOverdue(members: Member[]): number {
let count = 0;
for (const member of members) {
if (member.overdueDays > 30) {
sendReminder(member.name);
count += 1;
}
}
return count;
}
export function dailyReport(members: Member[]): string {
return `今日催还 ${remindOverdue(members)} 人`;
}
export const MEMBERS: Member[] = [
{ name: "小林", overdueDays: 45 },
{ name: "小陈", overdueDays: 3 },
{ name: "小吴", overdueDays: 31 },
];
console.log(dailyReport(MEMBERS));
console.log(sent.join(","));
// 输出:
// 今日催还 2 人
// 小林,小吴
import test, { beforeEach } from "node:test";
import assert from "node:assert/strict";
import { MEMBERS, dailyReport, sent } from "./target";
beforeEach(() => {
sent.length = 0;
});
test("日报写出催还人数", () => {
assert.equal(dailyReport(MEMBERS), "今日催还 2 人");
});
test("每人只提醒一次", () => {
dailyReport(MEMBERS);
assert.deepEqual(sent, ["小林", "小吴"]);
});
test("没人逾期", () => {
assert.equal(dailyReport([{ name: "小陈", overdueDays: 3 }]), "今日催还 0 人");
assert.deepEqual(sent, []);
});
  • 有返回值的函数不该有看得到的副作用。检验方法:连调两次和调一次结果是否一样。
  • 拆法:复制原函数做查询、删掉副作用,调用方改为“先查询、再调修改”,最后去掉原函数的返回值,让它复用查询。
  • 拆开之后,查询可以放心多调、单独测试,新调用方只想要答案时不会误触修改。
  • 返回值描述的就是修改结果(pop、插入后返回 ID),或需要原子性时,可以不拆。
  • 这条原则叫命令查询分离,Python 的 setdefault 是一个值得知道的例外。