以管道取代循环

以管道取代循环（Replace Loop with Pipeline）把一个手写的循环换成一串集合操作：筛选、映射、归约。 它解决的痛是：循环里筛选、转换、累加混在一起，读的人要在心里模拟每一轮迭代才知道它最后得到什么。

一个场景

小葛要从公司的办公室清单里找出印度的办公室，输出城市和电话。清单是一段 CSV 文本，第一行是表头。

原来的写法

"""办公室清单：重构前。"""

CSV = """office,country,telephone
Chicago,USA,+1 312 373 1000
Beijing,China,+86 4008 900 505
Bangalore,India,+91 80 4064 9570
Porto Alegre,Brazil,+55 51 3079 3550
Chennai,India,+91 44 660 44766
"""


def acquire_data(text: str) -> list[dict]:
    lines = text.splitlines()
    first_line = True
    result = []
    for line in lines:
        if first_line:
            first_line = False
            continue
        if line.strip() == "":
            continue
        record = line.split(",")
        if record[1].strip() == "India":
            result.append({"city": record[0].strip(), "phone": record[2].strip()})
    return result


if __name__ == "__main__":
    for row in acquire_data(CSV):
        print(row["city"], row["phone"])

# 输出：
# Bangalore +91 80 4064 9570
# Chennai +91 44 660 44766

// 办公室清单：重构前。
export const CSV = `office,country,telephone
Chicago,USA,+1 312 373 1000
Beijing,China,+86 4008 900 505
Bangalore,India,+91 80 4064 9570
Porto Alegre,Brazil,+55 51 3079 3550
Chennai,India,+91 44 660 44766
`;

export function acquireData(text: string): { city: string; phone: string }[] {
  const lines = text.split("\n");
  let firstLine = true;
  const result: { city: string; phone: string }[] = [];
  for (const line of lines) {
    if (firstLine) {
      firstLine = false;
      continue;
    }
    if (line.trim() === "") continue;
    const record = line.split(",");
    if (record[1].trim() === "India") {
      result.push({ city: record[0].trim(), phone: record[2].trim() });
    }
  }
  return result;
}

for (const row of acquireData(CSV)) console.log(row.city, row.phone);

// 输出：
// Bangalore +91 80 4064 9570
// Chennai +91 44 660 44766

测试：

import unittest

from target import CSV, acquire_data


class OfficesTest(unittest.TestCase):
    def test_finds_indian_offices(self):
        self.assertEqual(
            acquire_data(CSV),
            [{"city": "Bangalore", "phone": "+91 80 4064 9570"}, {"city": "Chennai", "phone": "+91 44 660 44766"}],
        )

    def test_header_only(self):
        self.assertEqual(acquire_data("office,country,telephone\n"), [])

    def test_blank_lines_ignored(self):
        self.assertEqual(acquire_data("h,h,h\n\nPune,India,123\n\n"), [{"city": "Pune", "phone": "123"}])


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

import test from "node:test";
import assert from "node:assert/strict";
import { CSV, acquireData } from "./target";

test("找出印度的办公室", () => {
  assert.deepEqual(acquireData(CSV), [
    { city: "Bangalore", phone: "+91 80 4064 9570" },
    { city: "Chennai", phone: "+91 44 660 44766" },
  ]);
});

test("只有表头", () => {
  assert.deepEqual(acquireData("office,country,telephone\n"), []);
});

test("忽略空行", () => {
  assert.deepEqual(acquireData("h,h,h\n\nPune,India,123\n\n"), [{ city: "Pune", phone: "123" }]);
});

问题在哪

如果现在要加一个需求：同时输出巴西的办公室，或者按城市名排序。在循环里加条件要非常小心：first_line 这个标志变量控制着“跳过表头”，continue 散在两个地方，result.append 藏在最深一层的 if 里。改一处要先在心里跑一遍整个循环，确认改动没碰到另外几件事。

这个循环实际上在做五件事：跳过第一行、去掉空行、按逗号拆分、筛选印度、取出城市和电话。五件事被一个循环和一个标志变量编织在一起，要读完整个循环体才拼得出这个清单。这就是第 2 章说的循环这种坏味道。

重构之后

"""办公室清单：重构后。"""

CSV = """office,country,telephone
Chicago,USA,+1 312 373 1000
Beijing,China,+86 4008 900 505
Bangalore,India,+91 80 4064 9570
Porto Alegre,Brazil,+55 51 3079 3550
Chennai,India,+91 44 660 44766
"""


def acquire_data(text: str) -> list[dict]:
    lines = text.splitlines()[1:]
    records = (
        [field.strip() for field in line.split(",")]
        for line in lines
        if line.strip()
    )
    return [{"city": r[0], "phone": r[2]} for r in records if r[1] == "India"]


if __name__ == "__main__":
    for row in acquire_data(CSV):
        print(row["city"], row["phone"])

# 输出：
# Bangalore +91 80 4064 9570
# Chennai +91 44 660 44766

// 办公室清单：重构后。
export const CSV = `office,country,telephone
Chicago,USA,+1 312 373 1000
Beijing,China,+86 4008 900 505
Bangalore,India,+91 80 4064 9570
Porto Alegre,Brazil,+55 51 3079 3550
Chennai,India,+91 44 660 44766
`;

export function acquireData(text: string): { city: string; phone: string }[] {
  return text
    .split("\n")
    .slice(1)
    .filter((line) => line.trim() !== "")
    .map((line) => line.split(",").map((field) => field.trim()))
    .filter((record) => record[1] === "India")
    .map((record) => ({ city: record[0], phone: record[2] }));
}

for (const row of acquireData(CSV)) console.log(row.city, row.phone);

// 输出：
// Bangalore +91 80 4064 9570
// Chennai +91 44 660 44766

TS 版从上往下读，每一行就是那五件事中的一件，和刚才列出的清单一一对应。first_line 标志变量不见了，slice(1) 一个词说清“跳过表头”。加“同时要巴西”只改一个 filter 的条件；加排序就在末尾接一个 sort，不碰前面任何一步。

Python 版用列表推导和生成器表达式写同样的管道。Python 没有像 JS 那样的链式数组方法，推导式就是它的管道写法：中间那个圆括号的生成器表达式负责“去空行、拆字段”，最后一个列表推导负责“筛选、取字段”。一个推导式里塞两三个以上的 for 和 if 就难读了，这时把每一步写成一个有名字的生成器变量，像上面的 records 那样。

调试管道和调试循环的方式不太一样。循环里可以在任意一行打断点、打印变量；管道是一条表达式，中间结果没有名字。最简单的办法是临时把管道拆开，把每一步的结果赋给一个变量再打印，确认之后再合回去。JS 里也可以在链中间插一个 .map((x) => { console.log(x); return x; })，不改变数据，只看一眼。Python 的推导式可以先改写成普通 for 循环调试，这也说明推导式写得太复杂时，本来就该拆开。

怎么一步步做

1. 新建一个变量，放循环要处理的集合，后面的管道操作都接在它上面。
2. 从循环顶部开始，把每一段行为逐个换成一个管道操作。跳过表头 → slice(1) 或 [1:]；去空行 → filter；拆字段 → map。每换一段跑一次测试。
3. 循环体里的条件判断变成 filter，对每个元素的转换变成 map，累加变成 reduce 或 sum。
4. 所有行为都换成管道之后，删掉循环。跑测试。

每一步都可以单独提交。如果某一步特别复杂，比如拆字段本身有十几行，先把它提炼成函数，管道里写 .map(parseRecord)。

Python 里还有 map() 和 filter() 两个内置函数，list(filter(lambda r: r[1] == "India", records)) 和推导式效果一样。社区的习惯是优先用推导式：条件和转换写在一行里，不需要 lambda，读起来更像自然语言。只有在已经有一个现成的函数可以直接传进去时，map(str.strip, fields) 这种写法才比推导式更短。

管道还改变了代码的“形状”：循环版本是一堆语句按顺序修改状态，读者要跟踪 first_line、result 在每一轮怎么变；管道版本是一个表达式，每一步都是“输入一个集合，输出一个集合”，没有中途被修改的变量。没有可变状态，就没有“这个变量此刻是什么值”的问题，这也是第 30 章、第 32 章反复在解决的事。

定义与适用场景

以管道取代循环：把对集合的迭代处理改写成一系列集合操作的组合，每个操作只做一件事，前一个操作的输出是后一个的输入。

它适合循环在做“筛选、转换、聚合”这类事情的时候，这也是绝大多数循环在做的事。常用的操作只有几个：filter（筛选）、map（转换每个元素）、reduce / sum / min / max（聚合成一个值）、sort（排序）、slice（取一段）、flatMap（展开）。Python 的推导式、sum、any、all、sorted、itertools，JS 的数组方法，都是这一套。

循环在做别的事时就不适合，比如需要提前退出（break）、需要同时修改外部状态、需要在迭代中改变集合本身。前两种有时能用 find、some、next 表达，第三种应该保留循环。

相关的设计模式

管道的每一步都在“逐个取出元素”，这正是迭代器（Iterator）模式要做的事：不暴露集合的内部结构，按顺序访问每个元素。Python 的生成器、JS 的迭代协议（Symbol.iterator）都是语言内置的迭代器，所以本章的管道不需要自己写迭代器类。Python 的生成器表达式还是惰性的：上面 records 那一步不会一次把所有行拆好，而是下游要一个才拆一个。本例的输入已经整段读进了内存，惰性的好处不明显；如果输入是逐行读取的文件对象，同样的写法就不必把整个文件放进内存。迭代器模式见《设计模式》书第 22 章。

JS 的数组方法是“急切”的：每调用一次 filter 或 map，就立刻遍历全部元素、生成一个新数组。数据量不大时这不是问题，读起来清楚最重要。处理很大或无限的数据流时，可以用生成器函数自己写惰性的步骤，或者使用较新的运行时提供的迭代器辅助方法，它们像 Python 的生成器表达式一样，要一个才算一个。本书的例子都是小数据，用数组方法。

真实系统里的例子

pandas 式的链式调用

数据分析里，pandas 的写法 df[df.region == "华东"].groupby("product").amount.sum() 就是一条管道。下面用纯 Python 写一个最小的 Frame 类模拟这种链式调用，计算华东区各产品的销售额。

"""用纯 Python 模拟 pandas 风格的链式调用。"""

from collections import defaultdict
from collections.abc import Callable


class Frame:
    def __init__(self, rows: list[dict]) -> None:
        self.rows = rows

    def filter_rows(self, predicate: Callable[[dict], bool]) -> "Frame":
        return Frame([r for r in self.rows if predicate(r)])

    def sum_by(self, key: str, value: str) -> dict:
        totals: dict = defaultdict(int)
        for r in self.rows:
            totals[r[key]] += r[value]
        return dict(sorted(totals.items()))


SALES = [
    {"region": "华东", "product": "咖啡豆", "amount": 300},
    {"region": "华北", "product": "咖啡豆", "amount": 200},
    {"region": "华东", "product": "手冲壶", "amount": 500},
    {"region": "华东", "product": "咖啡豆", "amount": 150},
]

if __name__ == "__main__":
    print(Frame(SALES).filter_rows(lambda r: r["region"] == "华东").sum_by("product", "amount"))

# 输出：
# {'咖啡豆': 450, '手冲壶': 500}

真实的 pandas 底层用向量化计算，比 Python 循环快得多；但从读代码的角度，它和本章的管道是同一种思路：一串名字清楚的步骤，每步只做一件事。

订单数组的筛选、映射、归约

前端拿到订单数组后算“本月已付款订单的总金额和平均金额”，用循环写要三个累加变量，用管道写是一条链。

// 订单数组：本月已付款订单的总额和均额。
interface Order {
  month: string;
  paid: boolean;
  amount: number;
}

const orders: Order[] = [
  { month: "2026-10", paid: true, amount: 120 },
  { month: "2026-10", paid: false, amount: 80 },
  { month: "2026-09", paid: true, amount: 200 },
  { month: "2026-10", paid: true, amount: 60 },
];

const amounts = orders.filter((o) => o.month === "2026-10" && o.paid).map((o) => o.amount);
const total = amounts.reduce((sum, a) => sum + a, 0);
console.log(total, amounts.length ? total / amounts.length : 0);

// 输出：
// 180 90

还有一种循环看起来能改成管道，其实不该改：每一轮都依赖上一轮的结果。比如逐行计算累计余额、根据前一个状态决定下一个状态，用 reduce 硬写出来往往比循环更难读。管道擅长的是“每个元素独立处理”的场景，元素之间有前后依赖时，一个清楚的循环加上好的变量名就是最好的写法。

什么时候别用

链太长也难读。一条管道超过五六步，或者某一步的 lambda 本身有好几行，就把中间结果存进有名字的变量，或者把复杂的步骤提炼成函数。管道的目标是让每一步一眼可懂，不是把所有东西写成一行。

对性能极其敏感、数据量又大的热循环，JS 的每个数组方法都会创建一个新数组，链很长时内存开销可以测出来。先测，确认是瓶颈再改回循环，或者换成惰性的迭代器写法。

容易混淆的手法

手法	它关心什么	以管道取代循环关心什么
拆分循环（第 26 章）	一个循环做多件事，拆成多个循环	一个循环里的步骤，改写成操作链；常在拆分循环之后做
以函数调用取代内联代码（第 24 章）	手写的逻辑换成一个已有函数	整个循环换成一串操作
拆分阶段（第 11 章）	函数级别的阶段划分	集合处理里的步骤划分；管道是更细粒度的阶段

怎么判断一条管道写得好？一个实用的标准是：把每一步的名字按顺序念出来，是否就是这段逻辑的中文描述。“去表头、去空行、拆字段、筛印度、取城市和电话”念出来就是需求本身，这条管道就是好的。如果念出来的是“map 一个 lambda，再 reduce 一个 lambda”，说不清每一步在干什么，就该给步骤起名字：把 lambda 换成有名字的函数，或者把中间结果放进有名字的变量。

练一练

下面的函数从访问日志里统计每个页面被成功访问（状态码 200）的次数，只算 /blog/ 开头的页面，结果按次数从多到少排列。用管道重写，输出不变。

"""练习：博客页面访问统计。"""

LOG = """GET /blog/a 200
GET /blog/b 404
GET /about 200
GET /blog/a 200
GET /blog/c 200
GET /blog/c 200
GET /blog/c 200
"""


def blog_hits(log_text: str) -> list[tuple[str, int]]:
    counts: dict[str, int] = {}
    for line in log_text.splitlines():
        parts = line.split()
        if len(parts) != 3:
            continue
        _, path, status = parts
        if status != "200":
            continue
        if not path.startswith("/blog/"):
            continue
        counts[path] = counts.get(path, 0) + 1
    result = list(counts.items())
    result.sort(key=lambda item: (-item[1], item[0]))
    return result


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

# 输出：
# [('/blog/c', 3), ('/blog/a', 2)]

import unittest

from target import LOG, blog_hits


class HitsTest(unittest.TestCase):
    def test_counts_and_order(self):
        self.assertEqual(blog_hits(LOG), [("/blog/c", 3), ("/blog/a", 2)])

    def test_ties_sorted_by_path(self):
        self.assertEqual(blog_hits("GET /blog/z 200\nGET /blog/y 200\n"), [("/blog/y", 1), ("/blog/z", 1)])

    def test_malformed_lines_skipped(self):
        self.assertEqual(blog_hits("坏行\nGET /blog/a 200\n"), [("/blog/a", 1)])


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

// 练习：博客页面访问统计。
export const LOG = `GET /blog/a 200
GET /blog/b 404
GET /about 200
GET /blog/a 200
GET /blog/c 200
GET /blog/c 200
GET /blog/c 200
`;

export function blogHits(logText: string): [string, number][] {
  const counts: Record<string, number> = {};
  for (const line of logText.split("\n")) {
    const parts = line.split(" ");
    if (parts.length !== 3) continue;
    const [, path, status] = parts;
    if (status !== "200") continue;
    if (!path.startsWith("/blog/")) continue;
    counts[path] = (counts[path] ?? 0) + 1;
  }
  const result = Object.entries(counts);
  result.sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0]));
  return result;
}

console.log(JSON.stringify(blogHits(LOG)));

// 输出：
// [["/blog/c",3],["/blog/a",2]]

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

test("计数和排序", () => {
  assert.deepEqual(blogHits(LOG), [
    ["/blog/c", 3],
    ["/blog/a", 2],
  ]);
});

test("次数相同按路径排", () => {
  assert.deepEqual(blogHits("GET /blog/z 200\nGET /blog/y 200\n"), [
    ["/blog/y", 1],
    ["/blog/z", 1],
  ]);
});

test("跳过格式不对的行", () => {
  assert.deepEqual(blogHits("坏行\nGET /blog/a 200\n"), [["/blog/a", 1]]);
});

最后提醒一句：改写成管道之后，测试一行都不用动，三个用例在两个版本上都通过。管道改变的是代码的写法，不是它的结果，这正是重构的定义。

小结

• 循环在做筛选、转换、聚合时，换成 filter、map、reduce 一类的操作链，每步只做一件事。
• 从循环顶部开始逐段替换，每换一段跑一次测试，最后删掉循环。
• Python 用推导式和生成器表达式写管道；步骤多时把中间结果放进有名字的变量。
• 需要 break、修改外部状态、边迭代边改集合的循环，保留原样。
• 管道是迭代器模式的日常用法；Python 的生成器还让它惰性求值。