拆分阶段

拆分阶段（Split Phase）把一段先后做两件事的代码拆成前后两个阶段，中间用一个数据结构传递结果。 它解决的痛是：两件事的代码缠在一起，改其中一件要在另一件的代码里找，想单独复用其中一件也办不到。

一个场景

小韩给一个批发商城写报价。报价分两步：先算商品价格，批量购买有折扣；再按商品原价选运费档位，加上运费。小韩把两步写在一个函数里，从上往下一气呵成。

原来的写法

"""报价：重构前。"""


def price_order(product: dict, quantity: int, shipping_method: dict) -> float:
    base_price = product["base_price"] * quantity
    discount = max(quantity - product["discount_threshold"], 0) * product["base_price"] * product["discount_rate"]
    price = base_price - discount
    if base_price > shipping_method["discount_threshold"]:
        shipping_per_case = shipping_method["discounted_fee"]
    else:
        shipping_per_case = shipping_method["fee_per_case"]
    shipping_cost = quantity * shipping_per_case
    return price + shipping_cost


PRODUCT = {"base_price": 10.0, "discount_threshold": 50, "discount_rate": 0.1}
SHIPPING = {"discount_threshold": 500, "discounted_fee": 1.0, "fee_per_case": 2.0}

if __name__ == "__main__":
    print(price_order(PRODUCT, 60, SHIPPING))
    print(price_order(PRODUCT, 20, SHIPPING))

# 输出：
# 650.0
# 240.0

// 报价：重构前。
export interface Product {
  basePrice: number;
  discountThreshold: number;
  discountRate: number;
}

export interface ShippingMethod {
  discountThreshold: number;
  discountedFee: number;
  feePerCase: number;
}

export function priceOrder(product: Product, quantity: number, shippingMethod: ShippingMethod): number {
  const basePrice = product.basePrice * quantity;
  const discount = Math.max(quantity - product.discountThreshold, 0) * product.basePrice * product.discountRate;
  const price = basePrice - discount;
  const shippingPerCase =
    basePrice > shippingMethod.discountThreshold ? shippingMethod.discountedFee : shippingMethod.feePerCase;
  const shippingCost = quantity * shippingPerCase;
  return price + shippingCost;
}

export const PRODUCT: Product = { basePrice: 10, discountThreshold: 50, discountRate: 0.1 };
export const SHIPPING: ShippingMethod = { discountThreshold: 500, discountedFee: 1, feePerCase: 2 };

console.log(priceOrder(PRODUCT, 60, SHIPPING));
console.log(priceOrder(PRODUCT, 20, SHIPPING));

// 输出：
// 650
// 240

测试：

import unittest

from target import PRODUCT, SHIPPING, price_order


class PricingTest(unittest.TestCase):
    def test_bulk_order_gets_discount_and_cheap_shipping(self):
        self.assertEqual(price_order(PRODUCT, 60, SHIPPING), 650.0)

    def test_small_order_pays_full_price_and_shipping(self):
        self.assertEqual(price_order(PRODUCT, 20, SHIPPING), 240.0)

    def test_shipping_discount_needs_strictly_more_than_threshold(self):
        self.assertEqual(price_order(PRODUCT, 50, SHIPPING), 600.0)


if __name__ == "__main__":
    unittest.main()

import test from "node:test";
import assert from "node:assert/strict";
import { PRODUCT, SHIPPING, priceOrder } from "./target";

test("批量订单有折扣和优惠运费", () => {
  assert.equal(priceOrder(PRODUCT, 60, SHIPPING), 650);
});

test("小订单全价加标准运费", () => {
  assert.equal(priceOrder(PRODUCT, 20, SHIPPING), 240);
});

test("运费优惠要严格超过门槛", () => {
  assert.equal(priceOrder(PRODUCT, 50, SHIPPING), 600);
});

问题在哪

如果现在要加一个需求：购物车页面要显示“预计商品价格”，不含运费，因为用户还没选配送方式。这个数字就是 price_order 前三行算出来的 price，但它被埋在函数中间，后面紧接着就是运费。要单独拿到它，只能把前三行复制到购物车模块，以后批量折扣的规则改了要改两处。

第二个问题：运费规则和定价规则会因为不同的原因、被不同的人改。物流部门调运费档位，销售部门调折扣。两条规则在同一个函数里交织，改运费的人要读懂折扣代码才敢动手，而且 base_price 这个中间值同时被折扣和运费两边用，改折扣算法时得小心别碰坏运费的输入。

第三个问题在测试。想测运费档位的边界，比如“商品原价正好等于门槛时按哪一档算”，得先凑出一个商品和数量，让定价阶段恰好算出那个金额，再看运费。测的是运费，大半精力花在构造定价输入上。运费规则每多一条，这样的测试就多一份。

用第 2 章的名字说，这是发散式变化：一个函数因为两种不相关的原因被修改。

重构之后

"""报价：重构后。"""

from dataclasses import dataclass


@dataclass(frozen=True)
class PriceData:
    base_price: float
    quantity: int
    discount: float

    @property
    def price(self) -> float:
        return self.base_price - self.discount


def calculate_pricing(product: dict, quantity: int) -> PriceData:
    base_price = product["base_price"] * quantity
    discount = max(quantity - product["discount_threshold"], 0) * product["base_price"] * product["discount_rate"]
    return PriceData(base_price, quantity, discount)


def apply_shipping(price_data: PriceData, shipping_method: dict) -> float:
    if price_data.base_price > shipping_method["discount_threshold"]:
        shipping_per_case = shipping_method["discounted_fee"]
    else:
        shipping_per_case = shipping_method["fee_per_case"]
    return price_data.price + price_data.quantity * shipping_per_case


def price_order(product: dict, quantity: int, shipping_method: dict) -> float:
    return apply_shipping(calculate_pricing(product, quantity), shipping_method)


PRODUCT = {"base_price": 10.0, "discount_threshold": 50, "discount_rate": 0.1}
SHIPPING = {"discount_threshold": 500, "discounted_fee": 1.0, "fee_per_case": 2.0}

if __name__ == "__main__":
    print(price_order(PRODUCT, 60, SHIPPING))
    print(price_order(PRODUCT, 20, SHIPPING))

# 输出：
# 650.0
# 240.0

// 报价：重构后。
export interface Product {
  basePrice: number;
  discountThreshold: number;
  discountRate: number;
}

export interface ShippingMethod {
  discountThreshold: number;
  discountedFee: number;
  feePerCase: number;
}

interface PriceData {
  readonly basePrice: number;
  readonly quantity: number;
  readonly discount: number;
}

function calculatePricing(product: Product, quantity: number): PriceData {
  const basePrice = product.basePrice * quantity;
  const discount = Math.max(quantity - product.discountThreshold, 0) * product.basePrice * product.discountRate;
  return { basePrice, quantity, discount };
}

function applyShipping(priceData: PriceData, shippingMethod: ShippingMethod): number {
  const shippingPerCase =
    priceData.basePrice > shippingMethod.discountThreshold ? shippingMethod.discountedFee : shippingMethod.feePerCase;
  return priceData.basePrice - priceData.discount + priceData.quantity * shippingPerCase;
}

export function priceOrder(product: Product, quantity: number, shippingMethod: ShippingMethod): number {
  return applyShipping(calculatePricing(product, quantity), shippingMethod);
}

export const PRODUCT: Product = { basePrice: 10, discountThreshold: 50, discountRate: 0.1 };
export const SHIPPING: ShippingMethod = { discountThreshold: 500, discountedFee: 1, feePerCase: 2 };

console.log(priceOrder(PRODUCT, 60, SHIPPING));
console.log(priceOrder(PRODUCT, 20, SHIPPING));

// 输出：
// 650
// 240

购物车那个需求现在是一行：calculate_pricing(product, quantity).price。定价和运费各自成了一个函数，销售改折扣只看 calculate_pricing，物流改档位只看 apply_shipping。两者之间唯一的接口是 PriceData：第二阶段需要什么，第一阶段就交出什么，写在一个结构里，一目了然。

运费的测试也简单了：直接构造一个 PriceData(base_price=500, quantity=50, discount=0) 传给 apply_shipping，不用再凑商品和数量。边界条件想测几个测几个。

注意 price_order 还在，签名没变，只是变成了两个阶段的拼接。测试一个字没改。

怎么一步步做

1. 把第二阶段的代码提炼成一个函数（第 4 章）。这里是从“选运费档位”开始到结尾那几行，提成 apply_shipping。此时它的参数是第一阶段算出来的所有中间变量：base_price、price、quantity。跑测试。
2. 建一个中间数据结构，先是空的。把它加为第二阶段函数的参数。跑测试。
3. 把第二阶段用到的每个中间变量，一个一个搬进中间结构：在第一阶段把值放进去，在第二阶段从结构里读，删掉对应的参数。每搬一个跑一次测试。price 这种能从别的字段算出来的，可以不存，做成结构的属性。
4. 把第一阶段的代码也提炼成函数，返回中间结构。跑测试。
5. 原函数现在只剩两个调用，留着它作为对外入口，或者让调用方直接用两个阶段。

每一步都保持原函数能跑，所以中间任何时候都可以停。IDE 的 Extract Function 和 Introduce Parameter 能完成第 1、2 步的机械部分（IDE 可自动完成）。

为什么先提炼第二阶段而不是第一阶段？因为第二阶段的参数列表会直接告诉你中间结构该有哪些字段：提炼 apply_shipping 时 IDE 自动生成的参数是 base_price、price、quantity，这就是 PriceData 的字段清单。先提炼第一阶段的话，你得自己猜后面需要什么。

定义与适用场景

拆分阶段：把处理不同事情的代码分到各自独立的阶段，前一个阶段的输出是后一个阶段的输入，用一个明确的中间数据结构连接。

它最典型的样子是“先解析，再处理”。命令行工具先解析参数再执行；编译器先词法分析、再语法分析、再生成代码；接口处理先把请求解析成领域对象，再做业务，再组装响应。凡是代码的前半段在处理输入的格式、后半段在处理业务含义，都值得拆。另一类是本章的例子，两段业务规则由不同的人维护、按不同的节奏变化。

拆分的信号是：函数里有一个中间变量，前半段只往里写、后半段只从里读；或者有一条注释说“到这里 X 算完了，下面开始算 Y”。

中间数据结构怎么设计？原则是只放第二阶段需要的东西。PriceData 里没有 product，因为运费阶段不需要知道商品是什么；它有 quantity，因为运费按件数算。如果你发现第二阶段需要第一阶段的全部输入，说明两个阶段的边界可能画错了，或者它们根本不该拆。中间结构做成不可变的（Python 的 frozen 数据类，TS 的 readonly 字段），第二阶段就不能回头改第一阶段的结果，数据只往一个方向流。

阶段不限于两个。CSV 导入拆成解析、校验、入库三段，每两段之间各有一个中间结构。拆到什么程度为止？每个阶段能用一句话说清它做什么、能单独测试，就够了。

相关的设计模式

拆出来的阶段串起来就是管道（Pipeline）：每个阶段接收上一阶段的输出，产出下一阶段的输入，阶段之间只通过数据结构说话。Express 和 Koa 的中间件、Unix 的 |、数据处理里的 ETL 都是这个形状。管道不是 GoF 模式，《设计模式》书第 30 章“走出 GoF”会讲它和中间件管道；GoF 里最接近的是责任链（Chain of Responsibility），区别是责任链的每一环可以决定不往下传，管道的每一段总是把结果交给下一段。责任链见《设计模式》书第 23 章。

真实系统里的例子

CSV 导入：解析、校验、入库

后台的批量导入功能常常是一个函数从读文件做到写数据库。拆成解析、校验、入库三个阶段后，每个阶段能单独测，校验规则改了不用碰解析，换数据库不用碰校验。下面用标准库的 csv 模块和一个列表模拟数据库。

"""模拟 CSV 导入：一个函数 vs 三个阶段。"""

import csv
import io

CSV_TEXT = "name,email,age\n张三,[email protected],30\n李四,not-an-email,25\n王五,[email protected],abc\n"


def import_v1(text: str) -> tuple[list[dict], list[str]]:
    saved, errors = [], []
    for row_number, row in enumerate(csv.DictReader(io.StringIO(text)), start=2):
        if "@" not in row["email"]:
            errors.append(f"第 {row_number} 行：邮箱格式不对")
            continue
        if not row["age"].isdigit():
            errors.append(f"第 {row_number} 行：年龄要是数字")
            continue
        saved.append({"name": row["name"], "email": row["email"], "age": int(row["age"])})
    return saved, errors


def parse(text: str) -> list[dict]:
    return [{"line": n, **row} for n, row in enumerate(csv.DictReader(io.StringIO(text)), start=2)]


def validate(rows: list[dict]) -> tuple[list[dict], list[str]]:
    valid, errors = [], []
    for row in rows:
        if "@" not in row["email"]:
            errors.append(f"第 {row['line']} 行：邮箱格式不对")
        elif not row["age"].isdigit():
            errors.append(f"第 {row['line']} 行：年龄要是数字")
        else:
            valid.append(row)
    return valid, errors


def save(rows: list[dict]) -> list[dict]:
    return [{"name": r["name"], "email": r["email"], "age": int(r["age"])} for r in rows]


def import_v2(text: str) -> tuple[list[dict], list[str]]:
    valid, errors = validate(parse(text))
    return save(valid), errors


if __name__ == "__main__":
    print(import_v1(CSV_TEXT) == import_v2(CSV_TEXT))
    saved, errors = import_v2(CSV_TEXT)
    print(saved)
    print(errors)

# 输出：
# True
# [{'name': '张三', 'email': '[email protected]', 'age': 30}]
# ['第 3 行：邮箱格式不对', '第 4 行：年龄要是数字']

validate 现在可以用一个手写的行列表来测，不用先造 CSV 文本；save 换成真正的数据库调用时，解析和校验一行不动。

Express 处理器：解析请求、计算、组装响应

Web 框架里的处理器天然是三个阶段：从请求里取出并校验参数，调用业务逻辑，把结果变成响应。下面模拟一个算运费的接口，handlerV1 一口气做完，handlerV2 拆成三步。

// 模拟 Express 处理器：解析请求 → 领域计算 → 组装响应。
interface Req {
  query: Record<string, string | undefined>;
}

interface Res {
  status: number;
  json: unknown;
}

function handlerV1(req: Req): Res {
  const weight = Number(req.query.weight);
  const country = req.query.country ?? "CN";
  if (!Number.isFinite(weight) || weight <= 0) return { status: 400, json: { error: "weight 必须是正数" } };
  const fee = country === "CN" ? 8 + Math.max(0, weight - 1) * 3 : 40 + Math.max(0, weight - 1) * 15;
  return { status: 200, json: { fee: fee.toFixed(2), currency: "CNY" } };
}

interface ShippingQuery {
  weight: number;
  country: string;
}

function parseQuery(req: Req): ShippingQuery | string {
  const weight = Number(req.query.weight);
  if (!Number.isFinite(weight) || weight <= 0) return "weight 必须是正数";
  return { weight, country: req.query.country ?? "CN" };
}

function shippingFee(query: ShippingQuery): number {
  const extra = Math.max(0, query.weight - 1);
  return query.country === "CN" ? 8 + extra * 3 : 40 + extra * 15;
}

function respond(fee: number): Res {
  return { status: 200, json: { fee: fee.toFixed(2), currency: "CNY" } };
}

function handlerV2(req: Req): Res {
  const parsed = parseQuery(req);
  if (typeof parsed === "string") return { status: 400, json: { error: parsed } };
  return respond(shippingFee(parsed));
}

const good: Req = { query: { weight: "2.5", country: "US" } };
const bad: Req = { query: { weight: "-1" } };
console.log(JSON.stringify(handlerV1(good)) === JSON.stringify(handlerV2(good)));
console.log(JSON.stringify(handlerV1(bad)) === JSON.stringify(handlerV2(bad)));
console.log(JSON.stringify(handlerV2(good)));

// 输出：
// true
// true
// {"status":200,"json":{"fee":"62.50","currency":"CNY"}}

shippingFee 接收的是干净的领域数据，不知道 HTTP 的存在，可以直接拿去给命令行工具或者定时任务复用。测它也只要构造一个 { weight: 2.5, country: "US" }，不用伪造请求对象。

Express 的中间件机制本身就是拆分阶段的产物：express.json() 这类中间件负责把请求体解析成对象挂在 req.body 上，后面的处理器只读解析好的结果。框架把“解析”这个阶段抽出来做成了可插拔的部件，每个处理器就不用再写一遍。

什么时候别用

两件事之间没有清晰的中间数据，或者中间数据和原始输入几乎一样，拆出来只是多了一层搬运。十行以内的函数，两个阶段各五行，拆不拆差别不大，先放着。

拆分的目的是让两个阶段独立变化，如果它们总是一起改、改的人也是同一个，那它们可能本来就是一件事。

拆过头的样子是：每个阶段只有两三行，阶段之间却定义了一套精致的中间类型，读主流程要在五个文件之间跳。中间结构的字段数应该明显少于原始输入，阶段的数量应该和“会独立变化的事情”的数量一样多，不多不少。

另一个常见的错误是让第二阶段拿到第一阶段的全部输入。比如 apply_shipping 的参数里又传了一遍 product，因为“万一以后要用”。这样两个阶段其实还是耦合的，只是中间多了一个结构。第二阶段真的需要某个原始字段，就把那个字段放进中间结构，而不是把整个输入透传过去。

容易混淆的手法

手法	它关心什么	拆分阶段关心什么
提炼函数（第 4 章）	给一段代码起名字	两段代码之间的数据流；拆分阶段用提炼函数做，但多了一个显式的中间结构
引入参数对象（第 9 章）	把散参数合成对象	中间结构也是一个参数对象，但它的来源是前一个阶段的输出而不是调用方
函数组合成变换（第 10 章）	一次算好派生值附在数据上	变换可以是拆分阶段里的第一阶段

练一练

下面的函数读一段日志，统计每个模块出错的次数，解析和统计写在一起。把它拆成两个阶段：parse_lines 把每行解析成一条记录，summarize 从记录里算出统计。error_summary 的签名和输出不变。

"""练习：错误日志统计。"""

LOG = """2026-10-07 10:00:01 ERROR db 连接超时
2026-10-07 10:00:05 INFO api 请求完成
2026-10-07 10:01:12 ERROR api 参数缺失
2026-10-07 10:02:30 ERROR db 连接超时
"""


def error_summary(log_text: str) -> str:
    counts: dict[str, int] = {}
    for line in log_text.splitlines():
        parts = line.split(" ", 4)
        if len(parts) < 5:
            continue
        level, module = parts[2], parts[3]
        if level == "ERROR":
            counts[module] = counts.get(module, 0) + 1
    return ", ".join(f"{module}: {counts[module]}" for module in sorted(counts))


if __name__ == "__main__":
    print(error_summary(LOG))

# 输出：
# api: 1, db: 2

import unittest

from target import LOG, error_summary


class LogTest(unittest.TestCase):
    def test_counts_errors_per_module(self):
        self.assertEqual(error_summary(LOG), "api: 1, db: 2")

    def test_ignores_non_error_lines(self):
        self.assertEqual(error_summary("2026-10-07 10:00:05 INFO api 请求完成\n"), "")

    def test_skips_malformed_lines(self):
        self.assertEqual(error_summary("坏行\n2026-10-07 10:00:01 ERROR db x\n"), "db: 1")


if __name__ == "__main__":
    unittest.main()

// 练习：错误日志统计。
export const LOG = `2026-10-07 10:00:01 ERROR db 连接超时
2026-10-07 10:00:05 INFO api 请求完成
2026-10-07 10:01:12 ERROR api 参数缺失
2026-10-07 10:02:30 ERROR db 连接超时
`;

export function errorSummary(logText: string): string {
  const counts: Record<string, number> = {};
  for (const line of logText.split("\n")) {
    const parts = line.split(" ");
    if (parts.length < 5) continue;
    const [, , level, module] = parts;
    if (level === "ERROR") counts[module] = (counts[module] ?? 0) + 1;
  }
  return Object.keys(counts)
    .sort()
    .map((module) => `${module}: ${counts[module]}`)
    .join(", ");
}

console.log(errorSummary(LOG));

// 输出：
// api: 1, db: 2

import test from "node:test";
import assert from "node:assert/strict";
import { LOG, errorSummary } from "./target";

test("按模块统计错误数", () => {
  assert.equal(errorSummary(LOG), "api: 1, db: 2");
});

test("忽略非错误行", () => {
  assert.equal(errorSummary("2026-10-07 10:00:05 INFO api 请求完成\n"), "");
});

test("跳过格式不对的行", () => {
  assert.equal(errorSummary("坏行\n2026-10-07 10:00:01 ERROR db x\n"), "db: 1");
});

小结

• 一个函数前半段做一件事、后半段做另一件事，中间有一个“算完了”的时刻，就拆成两个阶段。
• 两个阶段之间用一个明确的中间数据结构连接，第二阶段需要什么，第一阶段就交出什么。
• 先提炼第二阶段，再建中间结构，把中间变量一个一个搬进去，最后提炼第一阶段。
• 最典型的拆法是“解析输入”和“处理业务”分开，后者不再知道输入长什么样。
• 原函数可以留着当入口，签名不变，测试不用改。