02 · 算法设计范式全谱
“分治、贪心、DP、回溯——名字你都会背。但你知道什么时候该用哪个吗？“
开场：七个问题
老陈：我给你七个真实问题，你先凭直觉说，每个该用什么范式。
① 10 亿个数里找第 K 大 （K = 100）
② 安排会议室：100 个会议，最少用几个会议室
③ 背包问题：容量 50，20 个物品，最大化价值
④ 求解数独
⑤ 给 100 万个点，找最近的 K 个
⑥ 旅行商问题：20 个城市，最短路径
⑦ 海量数据里统计出现次数最多的 100 个词
小林：①排序或者堆……②贪心？③动态规划……④回溯……⑤……堆？⑥回溯？⑦堆？
老陈：你答对了 5 个，但②和⑤错了。而且更重要的是——你答不上来”为什么”。
小林：②不是贪心吗？按开始时间排序，然后依次安排……
老陈：会议室问题是贪心，但”按开始时间排序”是错的。 你自己想想为什么。
小林：……（想了一分钟）……如果一个会议 9 点开始 18 点结束，另一个 10 点开始 11 点结束，按开始时间排会把第一个排前面，但它占了整天。
老陈：对。正确的贪心策略不止一种，而且都不是”按开始时间排”。 这就是我要说的第一件事：知道”用贪心”只是第一步，知道”贪心策略是什么”才是关键。
小林：那⑤呢？
老陈：⑤经典的答案是堆，但在特定场景下有更好的解法。 你想想：如果这 100 万个点是二维平面上的，而且你要查询很多次呢？
小林：……建索引？KD 树？
老陈：对。这就是我要说的第二件事：算法的范式选择，取决于你的约束条件——数据规模、查询次数、是否允许近似、内存限制。没有脱离约束的”最好算法”。
范式一：分治（Divide and Conquer）
核心思想
1. 分解：把问题分成若干个子问题
2. 解决：递归解决子问题
3. 合并：把子问题的解合并成原问题的解
经典案例：归并排序
func MergeSort(arr []int) []int {
if len(arr) <= 1 {
return arr
}
mid := len(arr) / 2
left := MergeSort(arr[:mid])
right := MergeSort(arr[mid:])
return merge(left, right)
}
func merge(a, b []int) []int {
result := make([]int, 0, len(a)+len(b))
i, j := 0, 0
for i < len(a) && j < len(b) {
if a[i] <= b[j] {
result = append(result, a[i])
i++
} else {
result = append(result, b[j])
j++
}
}
result = append(result, a[i:]...)
result = append(result, b[j:]...)
return result
}复杂度：T(n) = 2T(n/2) + O(n) → O(n log n)（主定理）
案例 2：快速选择（问题 ①）
10 亿个数找第 K 大，K=100。
// 方案 A：排序 → O(n log n)，10 亿个数约 30 秒
// 方案 B：堆 → O(n log K)，log(100) ≈ 6.6 ★ 推荐
// 方案 C：快速选择 → O(n) 平均，但是原地，需要可读写的数组
// 方案 B：最小堆
type MinHeap []int
func (h MinHeap) Len() int { return len(h) }
func (h MinHeap) Less(i, j int) bool { return h[i] < h[j] }
func (h MinHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *MinHeap) Push(x any) { *h = append(*h, x.(int)) }
func (h *MinHeap) Pop() any {
old := *h
n := len(old)
x := old[n-1]
*h = old[:n-1]
return x
}
func TopK(nums []int, k int) []int {
h := make(MinHeap, 0, k)
for _, num := range nums {
if h.Len() < k {
heap.Push(&h, num)
} else if num > h[0] { // h[0] 是当前 Top-K 中最小的
h[0] = num // ★ 直接替换堆顶，比 Pop+Push 快
heap.Fix(&h, 0) // 下沉，O(log k)
}
}
// 转成降序输出
result := make([]int, h.Len())
for i := len(result) - 1; i >= 0; i-- {
result[i] = heap.Pop(&h).(int)
}
return result
}为什么堆比排序好？
n = 10 亿, K = 100
排序: O(n log n) = 10^9 × 30 = 3 × 10^10 次操作
堆: O(n log K) = 10^9 × 6.6 = 6.6 × 10^9 次操作 ★ 快 4.5 倍
而且堆只需要 O(K) = 100 个元素的内存！
排序需要 O(n) = 10 亿个元素的内存（或者原地排序但要可写）
老陈：这个例子展示了算法选择的核心权衡：
- 时间复杂度：O(n log K) vs O(n log n)
- 空间复杂度：O(K) vs O(n)
- 能否流式处理：堆可以（来一个处理一个），排序不行
如果你的数据是流式的（比如日志流），堆是唯一选择。
案例 3：分治的杀手级应用 —— 最近点对
问题：平面上 n 个点，找距离最近的两个点。
- 朴素：O(n²)
- 分治：O(n log n)
type Point struct{ X, Y float64 }
func dist(a, b Point) float64 {
dx, dy := a.X-b.X, a.Y-b.Y
return math.Sqrt(dx*dx + dy*dy)
}
// 按 X 排序后的点集，用分治求最近距离
func ClosestPair(px []Point) (Point, Point, float64) {
if len(px) <= 3 {
return bruteForce(px)
}
mid := len(px) / 2
midPoint := px[mid]
// 分治
p1, q1, d1 := ClosestPair(px[:mid])
p2, q2, d2 := ClosestPair(px[mid:])
var p, q Point
d := d1
if d2 < d1 {
p, q, d = p2, q2, d2
}
// ★ 关键：合并步骤
// 如果最近点对跨越了中线，两个点一定在 [midX-d, midX+d] 的带状区域内
strip := []Point{}
for _, pt := range px {
if math.Abs(pt.X-midPoint.X) < d {
strip = append(strip, pt)
}
}
// 按 Y 排序（这里可以优化成预排序，避免每次排序）
sort.Slice(strip, func(i, j int) bool { return strip[i].Y < strip[j].Y })
// ★ 关键洞察：strip 里每个点只需要检查后面 7 个点
// 证明：在 d×2d 的矩形里，最多能放 8 个距离 >= d 的点
for i := 0; i < len(strip); i++ {
for j := i + 1; j < len(strip) && strip[j].Y-strip[i].Y < d; j++ {
if dd := dist(strip[i], strip[j]); dd < d {
d = dd
p, q = strip[i], strip[j]
}
}
}
return p, q, d
}
func bruteForce(pts []Point) (Point, Point, float64) {
minD := math.Inf(1)
var bp, bq Point
for i := 0; i < len(pts); i++ {
for j := i + 1; j < len(pts); j++ {
if d := dist(pts[i], pts[j]); d < minD {
minD, bp, bq = d, pts[i], pts[j]
}
}
}
return bp, bq, minD
}为什么是 O(n log n)？
T(n) = 2T(n/2) + O(n) ← 合并步骤是 O(n)（因为只检查 7 个邻居）
= O(n log n) ← 主定理
关键洞察：合并步骤看似要 O(n²)（strip 里两两比较），但因为”每个点只需检查后面 7 个”，实际是 O(n)。
老陈：这是分治最精妙的地方：分治不只是”把问题变小”，更重要的是”分解后能剪掉大量无用计算”。
这个”只检查 7 个”的证明，是整个算法的灵魂。 它不是编程技巧，是几何洞察。
记住这个模式：O(n²) → O(n log n) 的关键，往往在于”用某种结构性质证明大部分比较是不必要的”。
范式二：贪心（Greedy）
核心思想
每一步都做当前看来最好的选择，寄希望于全局最优。
贪心能成立需要两个条件：
- 贪心选择性质：局部最优选择能导致全局最优
- 最优子结构：问题的最优解包含子问题的最优解
注意：贪心不是”随便选一个策略”，而是”证明某个策略是正确的”。大多数贪心策略是错的。
案例 1：会议室问题（问题 ②）
错误示范：按开始时间排序
会议 A: 9:00 - 18:00
会议 B: 10:00 - 11:00
会议 C: 11:00 - 12:00
按开始时间排：A(9点) → B 和 C 要等，需要 3 个会议室？
实际：B、C 可以用同一个，A 用一个 → 2 个会议室
正确策略 1：按结束时间排序 + 优先队列
type Meeting struct{ Start, End int }
func MinRooms(meetings []Meeting) int {
if len(meetings) == 0 {
return 0
}
// 按开始时间排序
sort.Slice(meetings, func(i, j int) bool {
return meetings[i].Start < meetings[j].Start
})
// 最小堆，存当前占用的会议室的结束时间
h := &MinHeap{}
heap.Init(h)
heap.Push(h, meetings[0].End)
for i := 1; i < len(meetings); i++ {
m := meetings[i]
// 如果最早结束的会议室已经空出来了，复用它
if (*h)[0] <= m.Start {
heap.Pop(h)
}
heap.Push(h, m.End)
}
return h.Len()
}复杂度：O(n log n)（排序） + O(n log n)（堆操作）
正确策略 2：扫描线（更优雅，且能处理无限会议室）
// 差分数组思想
func MinRoomsSweep(meetings []Meeting, maxTime int) int {
diff := make([]int, maxTime+2)
for _, m := range meetings {
diff[m.Start]++
diff[m.End]--
}
cur, max := 0, 0
for t := 0; t <= maxTime; t++ {
cur += diff[t]
if cur > max {
max = cur
}
}
return max
}
// 如果时间范围很大（比如是时间戳），用排序代替数组
func MinRoomsSweepSorted(meetings []Meeting) int {
events := make([]struct{ Time, Delta int }, 0, len(meetings)*2)
for _, m := range meetings {
events = append(events, struct{ Time, Delta int }{m.Start, 1})
events = append(events, struct{ Time, Delta int }{m.End, -1})
}
// 同一时刻，结束事件要排在开始事件前面（房间可以立即复用）
sort.Slice(events, func(i, j int) bool {
if events[i].Time != events[j].Time {
return events[i].Time < events[j].Time
}
return events[i].Delta < events[j].Delta
})
cur, max := 0, 0
for _, e := range events {
cur += e.Delta
if cur > max {
max = cur
}
}
return max
}老陈：扫描线是一个通用技巧，不只是会议室问题。
想想这些问题的共同点：
- 会议室：区间重叠的最大值
- 股票买卖：区间和的最大值
- 天际线问题：区间高度的合并
- 公交车站：同时在车上的最多人数
它们都是”区间上的统计”，都可以用”事件 + 扫描线”解决。
这就是模式的价值：你不是记住了 N 个问题的解法，而是识别出了它们属于同一个模式。
案例 2：贪心失效的经典 —— 零钱兑换
// 贪心：每次选最大面额
func CoinChangeGreedy(coins []int, amount int) int {
sort.Sort(sort.Reverse(sort.IntSlice(coins)))
count := 0
for _, c := range coins {
count += amount / c
amount %= c
}
if amount != 0 {
return -1
}
return count
}什么时候贪心对？
coins = [25, 10, 5, 1]（美元） → 贪心总是最优 ✅
coins = [25, 20, 10, 5, 1] → amount=40: 贪心给 25+10+5=3枚
最优是 20+20=2枚 ❌
判断标准：只有当硬币系统是”规范的”（canonical）时贪心才成立。没有简单的充要条件，需要对具体面额系统做验证。
老陈：这是贪心最重要的教训：贪心的正确性需要证明，不能想当然。
大多数”看起来可以贪心”的问题，其实都不行。 当你想用贪心时，先试着构造一个反例。构造不出来，再尝试证明。
贪心的正确性证明：交换论证
活动选择问题（选最多的不重叠会议）：
贪心策略：每次选结束时间最早的。
证明（交换论证）：
设贪心选择是活动 g（结束最早），最优解是 S = {a1, a2, ...}
① a1 的结束时间 >= g 的结束时间（因为 g 结束最早）
② 把 S 里的 a1 换成 g，得到 S' = {g, a2, ...}
③ S' 仍然合法：因为 g 结束比 a1 早，所以 g 不会跟 a2 冲突
④ S' 的大小 = S 的大小
⑤ 所以存在一个包含贪心选择的最优解
⑥ 递归地，每一步的贪心选择都是最优的一部分
∎
这个证明模式叫”交换论证”，是证明贪心正确性的标准方法。
范式三：动态规划（Dynamic Programming）
核心思想
1. 定义状态：dp[i] 或 dp[i][j] 表示什么
2. 状态转移方程：dp[i] 怎么从更小的状态推出来
3. 边界条件
4. 计算顺序（保证算 dp[i] 时，依赖的状态已经算好）
适用条件：
- 最优子结构：问题的最优解包含子问题的最优解
- 重叠子问题：子问题会被重复计算（这是 DP 区别于分治的关键）
案例 1：0-1 背包（问题 ③）
// 二维 DP
func Knapsack2D(weights, values []int, capacity int) int {
n := len(weights)
dp := make([][]int, n+1)
for i := range dp {
dp[i] = make([]int, capacity+1)
}
for i := 1; i <= n; i++ {
for w := 0; w <= capacity; w++ {
// 不选第 i 个物品
dp[i][w] = dp[i-1][w]
// 选第 i 个物品
if w >= weights[i-1] {
if v := dp[i-1][w-weights[i-1]] + values[i-1]; v > dp[i][w] {
dp[i][w] = v
}
}
}
}
return dp[n][capacity]
}
// 空间优化：一维 DP（滚动数组）
// ★ 关键：容量要从大到小遍历，否则会重复选同一个物品
func Knapsack1D(weights, values []int, capacity int) int {
dp := make([]int, capacity+1)
for i := 0; i < len(weights); i++ {
for w := capacity; w >= weights[i]; w-- { // ★ 逆序！
if v := dp[w-weights[i]] + values[i]; v > dp[w] {
dp[w] = v
}
}
}
return dp[capacity]
}为什么一维时要逆序？
正序（错误）：
w = 5: dp[5] = max(dp[5], dp[5-2]+3) ← dp[3] 可能被本轮更新过
w = 6: dp[6] = max(dp[6], dp[6-2]+3) ← 用到了刚更新的 dp[4]
→ 相当于同一个物品被选了多次（变成完全背包）
逆序（正确）：
w = 10: dp[10] 用到 dp[8]（上一轮的）
w = 9: dp[9] 用到 dp[7]（上一轮的）
→ 每个物品只被考虑一次
老陈：这个”正序 vs 逆序”的区别，是 0-1 背包和完全背包的分界线。
- 0-1 背包（每个物品只能选一次）：容量逆序遍历
- 完全背包（每个物品可以选无限次）：容量正序遍历
就这一个遍历方向的差别。 很多人背下来了但不知道为什么——现在你知道了：逆序保证 dp[w-weight] 是”上一轮”的值，正序让它可以是”本轮”的值。
案例 2：最长公共子序列（LCS）
func LCS(a, b string) int {
m, n := len(a), len(b)
dp := make([][]int, m+1)
for i := range dp {
dp[i] = make([]int, n+1)
}
for i := 1; i <= m; i++ {
for j := 1; j <= n; j++ {
if a[i-1] == b[j-1] {
dp[i][j] = dp[i-1][j-1] + 1
} else {
dp[i][j] = max(dp[i-1][j], dp[i][j-1])
}
}
}
return dp[m][n]
}
func max(a, b int) int {
if a > b {
return a
}
return b
}状态转移的直觉：
如果 a[i-1] == b[j-1]:
两个字符匹配，LCS 长度 +1
dp[i][j] = dp[i-1][j-1] + 1
否则:
要么 a[i-1] 不在 LCS 里 → dp[i-1][j]
要么 b[j-1] 不在 LCS 里 → dp[i][j-1]
dp[i][j] = max(两者)
应用：diff 工具（git diff）的核心就是 LCS。
案例 3：DP 的优化 —— 从 O(n²) 到 O(n)
问题：最大子数组和（Kadane 算法）
// 朴素 DP: O(n²)
func MaxSubArrayNaive(nums []int) int {
n := len(nums)
maxSum := nums[0]
for i := 0; i < n; i++ {
sum := 0
for j := i; j < n; j++ {
sum += nums[j]
if sum > maxSum {
maxSum = sum
}
}
}
return maxSum
}
// Kadane: O(n)，本质是 DP + 状态压缩
func MaxSubArray(nums []int) int {
// dp[i] = 以 nums[i] 结尾的最大子数组和
// dp[i] = max(nums[i], dp[i-1] + nums[i])
// ↑ 要么从自己开始，要么接在前面
curMax := nums[0] // dp[i-1]
globalMax := nums[0]
for i := 1; i < len(nums); i++ {
curMax = max(nums[i], curMax+nums[i])
globalMax = max(globalMax, curMax)
}
return globalMax
}优化的本质：dp[i] 只依赖 dp[i-1]，所以不需要数组，一个变量就够。
这是 DP 空间优化的标准套路：
dp[i]依赖dp[i-1]→ O(1) 空间（滚动变量）dp[i][j]依赖dp[i-1][*]→ O(n) 空间（滚动数组）dp[i][j]依赖dp[i-1][j-1]和dp[i-1][j]→ O(n)，但要注意遍历顺序
范式四：回溯与分支限界
核心思想
回溯 = 暴力搜索 + 剪枝
func Backtrack(路径, 选择列表) {
if 满足结束条件 {
记录结果
return
}
for 选择 := range 选择列表 {
if 不满足约束（剪枝） {
continue
}
做选择
Backtrack(新路径, 新选择列表)
撤销选择 // ★ 回溯的核心
}
}案例 1：数独求解（问题 ④）
type Sudoku [9][9]byte // 0 表示空
func SolveSudoku(board *Sudoku) bool {
var empty [][2]int
for i := 0; i < 9; i++ {
for j := 0; j < 9; j++ {
if board[i][j] == 0 {
empty = append(empty, [2]int{i, j})
}
}
}
return backtrack(board, empty, 0)
}
func backtrack(board *Sudoku, empty [][2]int, idx int) bool {
if idx == len(empty) {
return true // 全部填完
}
i, j := empty[idx][0], empty[idx][1]
// ★ 优化：按"候选数最少的格子优先"填（MRV 启发式）
// 这里简化，直接按顺序
for v := byte(1); v <= 9; v++ {
if isValid(board, i, j, v) {
board[i][j] = v // 做选择
if backtrack(board, empty, idx+1) {
return true
}
board[i][j] = 0 // 撤销选择
}
}
return false
}
func isValid(board *Sudoku, row, col int, v byte) bool {
// 检查行
for c := 0; c < 9; c++ {
if board[row][c] == v {
return false
}
}
// 检查列
for r := 0; r < 9; r++ {
if board[r][col] == v {
return false
}
}
// 检查 3×3
br, bc := row/3*3, col/3*3
for r := br; r < br+3; r++ {
for c := bc; c < bc+3; c++ {
if board[r][c] == v {
return false
}
}
}
return true
}朴素回溯的问题：最坏情况 9^81 次尝试（虽然是天文数字，但实际因为约束强，速度还行）。
优化 1：MRV 启发式（Minimum Remaining Values）
// 每次选"候选数最少"的格子来填
// 这样能尽早发现矛盾，剪掉大量分支
func selectCell(board *Sudoku, empty [][2]int, used []bool) int {
bestIdx, minCandidates := -1, 10
for idx, pos := range empty {
if used[idx] {
continue
}
i, j := pos[0], pos[1]
count := 0
for v := byte(1); v <= 9; v++ {
if isValid(board, i, j, v) {
count++
}
}
if count < minCandidates {
minCandidates = count
bestIdx = idx
if count == 0 {
break // 有格子无解，立即失败
}
}
}
return bestIdx
}实测：MRV 能把求解时间从秒级降到毫秒级。
优化 2：位运算加速
// 用位掩码记录每行/列/宫已用的数字
type SudokuFast struct {
rows [9]uint16
cols [9]uint16
boxs [9]uint16
board [9][9]byte
}
// 候选集 = 全集 XOR (行已用 | 列已用 | 宫已用)
func (s *SudokuFast) candidates(r, c int) uint16 {
used := s.rows[r] | s.cols[c] | s.boxs[r/3*3+c/3]
return ^used & 0x1FF // 0x1FF = 9 个 1
}老陈：数独是回溯的完美教学案例，因为它有两种截然不同的优化思路：
① 减少搜索空间（算法层面）：MRV、约束传播 ② 加速单次判断（实现层面）：位运算、预计算
真实的搜索问题，往往需要两者结合。而且通常优化①的收益更大。
案例 2：分支限界（Branch and Bound）
和回溯的区别：
- 回溯：找到所有可行解
- 分支限界：找最优解，用”上下界”剪掉不可能更优的分支
问题：旅行商问题（TSP，问题 ⑥）
type TSP struct {
n int
dist [][]int // 距离矩阵
best int // 当前最优解
bestPath []int
}
// 分支限界 + 贪心下界
func (t *TSP) Solve() (int, []int) {
t.best = math.MaxInt32
visited := make([]bool, t.n)
path := []int{0}
visited[0] = true
t.dfs(0, 0, path, visited)
return t.best, t.bestPath
}
func (t *TSP) dfs(city, cost int, path []int, visited []bool) {
// ★ 剪枝 1：如果当前代价 + 乐观估计 >= 已知最优，剪掉
if cost+t.lowerBound(city, visited) >= t.best {
return
}
// 走完了
if len(path) == t.n {
total := cost + t.dist[city][0] // 回到起点
if total < t.best {
t.best = total
t.bestPath = append([]int{}, path...)
}
return
}
for next := 0; next < t.n; next++ {
if visited[next] {
continue
}
visited[next] = true
path = append(path, next)
t.dfs(next, cost+t.dist[city][next], path, visited)
path = path[:len(path)-1]
visited[next] = false
}
}
// 下界估计：从当前城市出发，加上"每个未访问城市的最小出边"之和
// ★ 这个下界必须"乐观"（一定小于真实最优解），否则会剪掉正确答案
func (t *TSP) lowerBound(city int, visited []bool) int {
bound := 0
for i := 0; i < t.n; i++ {
if visited[i] {
continue
}
minEdge := math.MaxInt32
for j := 0; j < t.n; j++ {
if i != j && !visited[j] && t.dist[i][j] < minEdge {
minEdge = t.dist[i][j]
}
}
if minEdge < math.MaxInt32 {
bound += minEdge
}
}
return bound
}复杂度：最坏仍是 O(n!)，但剪枝后实际能解决 n=20-30 的 TSP。
老陈：TSP 是 NP-完全问题，不存在多项式时间的精确算法。 分支限界是”最实用的精确算法”，但规模一大就不行了。
这时候你有三个选择：
- 近似算法：保证解在最优解的某个倍数内（如 2-近似、1.5-近似）
- 启发式：不保证质量，但实际很快（如遗传算法、模拟退火、蚁群）
- 问题特化：利用问题的特殊结构（如度量 TSP 有 1.5-近似的 Christofides 算法）
选择哪个，取决于你的业务约束。这就是”工程判断”。
范式五：随机化算法
核心思想
用随机性换取简单性或速度。
两类：
- Monte Carlo：可能给出错误答案，但概率可控
- Las Vegas：一定给出正确答案，但运行时间随机
案例 1：随机化快速排序（避免最坏情况）
func QuickSortRandom(a []int) {
if len(a) <= 12 {
insertionSort(a)
return
}
// ★ 随机选 pivot，避免有序输入退化到 O(n²)
pivotIdx := rand.Intn(len(a))
a[pivotIdx], a[len(a)-1] = a[len(a)-1], a[pivotIdx]
pivot := a[len(a)-1]
i := 0
for j := 0; j < len(a)-1; j++ {
if a[j] < pivot {
a[i], a[j] = a[j], a[i]
i++
}
}
a[i], a[len(a)-1] = a[len(a)-1], a[i]
QuickSortRandom(a[:i])
QuickSortRandom(a[i+1:])
}为什么随机化有用？
固定选第一个元素做 pivot：
· 已排序输入 → 每次分割是 1 : n-1 → O(n²)
· 这是"可以预测的最坏情况"，攻击者可以构造这样的输入
随机选 pivot：
· 期望时间复杂度 O(n log n)，与输入分布无关
· 没有"特定的最坏输入"，只有"运气不好"
老陈：随机化最重要的价值不是”更快”，而是”消除可预测性”。
这有安全意义：如果攻击者能构造触发最坏情况的输入，就能发起 算法复杂度攻击（Algorithmic Complexity Attack）。
真实案例：
- 早期的哈希表用固定哈希函数，攻击者可以构造大量哈希碰撞的 key，把 O(1) 的查找变成 O(n)，造成 DoS
- Go 的 map 用随机哈希种子（每个 map 实例不同），就是防御这个攻击
- Python、Java、PHP 都做过类似的修复
这是一个”算法选择影响系统安全”的例子——面试不会考，但生产环境会咬你。
案例 2：蓄水池抽样（Reservoir Sampling）
问题：从未知长度的数据流中，等概率抽取 k 个元素。
// 数据流长度未知，只能遍历一次，要等概率抽 k 个
func ReservoirSample(stream <-chan int, k int) []int {
reservoir := make([]int, 0, k)
count := 0
for item := range stream {
count++
if len(reservoir) < k {
// 前 k 个直接放入
reservoir = append(reservoir, item)
} else {
// 第 i 个元素（i > k），以 k/i 的概率替换蓄水池里的随机一个
j := rand.Intn(count)
if j < k {
reservoir[j] = item
}
}
}
return reservoir
}为什么这样是等概率的？
归纳证明：
假设处理完前 i-1 个元素时，每个元素在蓄水池中的概率是 k/(i-1)
处理第 i 个元素时：
① 第 i 个被选中的概率 = k/i ✓ 正确
② 之前在蓄水池里的元素 x 被保留的概率
= P(第 i 个没被选中) + P(第 i 个被选中且没替换 x)
= (1 - k/i) + (k/i) × (1 - 1/k)
= (i-k)/i + (k/i) × (k-1)/k
= (i-k)/i + (k-1)/i
= (i-k+k-1)/i
= (i-1)/i
③ 所以 x 仍在蓄水池的概率 = k/(i-1) × (i-1)/i = k/i ✓ 正确
∎
应用：
- 数据库的随机采样（不 COUNT 全表）
- 流式日志的抽样监控
- 分布式系统的数据采样
范式六：近似算法
核心思想
当问题是 NP-难时，放弃”最优解”，追求”保证质量的近似解”。
近似比（Approximation Ratio）：
如果算法给出的解 ≤ α × 最优解（最小化问题），称 α-近似
案例：顶点覆盖问题的 2-近似
问题：给定图，找最小的顶点集合，使得每条边至少有一个端点在集合中。（NP-难）
// 2-近似算法：随机选一条边，把两个端点都加入覆盖集，删除所有相邻的边，重复
func VertexCover2Approx(edges [][2]int, numVertices int) []int {
covered := make([]bool, numVertices)
remaining := make([][2]int, len(edges))
copy(remaining, edges)
for len(remaining) > 0 {
// 任取一条边
e := remaining[0]
u, v := e[0], e[1]
// 两个端点都加入覆盖集
covered[u] = true
covered[v] = true
// 删除所有与 u 或 v 相邻的边
newRemaining := remaining[:0]
for _, ee := range remaining {
if ee[0] != u && ee[0] != v && ee[1] != u && ee[1] != v {
newRemaining = append(newRemaining, ee)
}
}
remaining = newRemaining
}
result := []int{}
for i, c := range covered {
if c {
result = append(result, i)
}
}
return result
}为什么是 2-近似？
设算法选了 m 条边，覆盖集大小 = 2m
最优解至少要覆盖这 m 条边。这 m 条边两两不相邻（因为选完就删掉了相邻的），
所以最优解对每条边至少要选 1 个不同的顶点，即 OPT >= m
近似比 = 2m / OPT <= 2m / m = 2 ∎
老陈：这个证明极其简洁，但它揭示了近似算法的核心技巧——找一个”最优解的下界”，然后证明算法解不超过这个下界的 α 倍。
这里的下界是”选出的边数 m”，因为那些边两两不相交。
好的近似算法证明，往往就靠这样一个巧妙的观察。
范式七：外部排序与流式算法
问题 ⑦：海量数据 Top-K
约束：数据有 100GB，内存只有 2GB。
方案 1：堆（内存够放 K 个元素时）
// O(n log K) 时间，O(K) 空间
// 如果 K = 100 万，堆要 8MB，完全可行方案 2：外部排序（K 很大时）
// 1. 分块：每次读 1GB 到内存，排序，写到临时文件
// 2. 归并：K 路归并所有临时文件
func ExternalSort(inputPath, outputPath string, chunkSize int) error {
// 阶段 1：分块排序
var tempFiles []string
for {
chunk, err := readChunk(inputPath, chunkSize)
if err == io.EOF {
break
}
sort.Ints(chunk)
tempPath := writeTempFile(chunk)
tempFiles = append(tempFiles, tempPath)
}
// 阶段 2：K 路归并（用最小堆）
return kWayMerge(tempFiles, outputPath)
}方案 3：Count-Min Sketch（允许近似时）
// 用极小的内存估算频率
// 适合"不需要精确结果，只要大致排名"的场景
type CountMinSketch struct {
width int
depth int
table [][]uint32
hashes []func(string) uint32
}
func NewCMS(width, depth int) *CountMinSketch {
cms := &CountMinSketch{
width: width,
depth: depth,
table: make([][]uint32, depth),
hashes: make([]func(string) uint32, depth),
}
for i := 0; i < depth; i++ {
cms.table[i] = make([]uint32, width)
seed := uint32(i*2654435761 + 1) // 每个哈希函数不同的种子
cms.hashes[i] = func(s string) uint32 {
h := seed
for _, c := range s {
h = h*uint32(c) + seed
}
return h % uint32(width)
}
}
return cms
}
func (c *CountMinSketch) Add(key string, count uint32) {
for i := 0; i < c.depth; i++ {
c.table[i][c.hashes[i](key)] += count
}
}
// 返回估计值（可能偏大，不会偏小）
func (c *CountMinSketch) Estimate(key string) uint32 {
min := uint32(math.MaxUint32)
for i := 0; i < c.depth; i++ {
if v := c.table[i][c.hashes[i](key)]; v < min {
min = v
}
}
return min
}为什么叫”Count-Min”：估计时取所有哈希槽的最小值。
误差分析：
- 只会高估，不会低估（因为哈希碰撞会让计数变大）
- 误差范围：以概率 1-δ，误差不超过 ε×N（N 是总计数）
- 参数选择：width = e/ε，depth = ln(1/δ)
空间对比：
| 方案 | 空间 | 精确性 |
|---|---|---|
| 哈希表 | O(不同 key 数) | 精确 |
| 堆（Top-K） | O(K) | 精确 |
| Count-Min Sketch | O(width × depth) ≈ 几 KB | 近似，只高估 |
100GB 数据，1 亿个不同词：
- 哈希表：约 8GB（放不下！）
- Count-Min Sketch（ε=0.001, δ=0.01）：约 30KB ✓
老陈：这就是”用精确性换空间”的极致。
判断标准：你的业务能接受多大的误差？
- 统计热门搜索词 → 可以近似
- 统计用户账户余额 → 绝对不行
理解业务约束，比记住算法更重要。
七个问题的完整答案
| # | 问题 | 范式 | 复杂度 | 备注 |
|---|---|---|---|---|
| ① | 10 亿数找 Top-100 | 堆 / 快速选择 | O(n log K) | 堆可流式，快选更快但要可写数组 |
| ② | 会议室 | 扫描线 / 堆 | O(n log n) | ❌ 不是”按开始时间排序” |
| ③ | 0-1 背包 | 动态规划 | O(n×W) | 一维优化要逆序遍历 |
| ④ | 数独 | 回溯 + MRV 剪枝 | 指数（实际很快） | MRV 能把秒级降到毫秒级 |
| ⑤ | 最近 K 个点 | 堆 / KD 树 | O(n log K) / O(log n) 每次查询 | 多次查询用 KD 树 |
| ⑥ | TSP | 分支限界 / 近似 | O(n!) 最坏 | n>30 要用启发式 |
| ⑦ | 海量 Top-K | 堆 / 外部排序 / CMS | O(n log K) / O(n log n) / O(1) 空间 | 允许近似用 CMS |
思考题 ·【创造层】
你的公司要做一个”实时推荐系统”，需求：从 1000 万用户中，为每个用户实时找出”最相似的 10 个用户”。相似度用向量余弦距离计算，每个用户的特征向量是 128 维。QPS 1 万。请设计系统，说明你的算法选择和每一个权衡。
参考答案
先算账
数据量：1000 万用户 × 128 维 × 4 字节 = 5.12 GB
QPS: 1 万
朴素方案：对每个请求，计算与 1000 万用户的余弦相似度
= 1 万 × 1000 万 × 128 = 1.28 × 10^13 次浮点运算/秒
= 12.8 TFLOPS
★ 这需要几十台高端 GPU 才能跑得动。不可行。
结论：暴力计算不可行，必须用近似算法或索引。
方案对比
方案 A：暴力 + 堆（基线）
func TopKSimilar(query []float32, users [][]float32, k int) []int {
// 最小堆存 Top-K
h := make(SimilarityHeap, 0, k)
for i, u := range users {
sim := cosine(query, u)
if h.Len() < k {
heap.Push(&h, Item{i, sim})
} else if sim > h[0].sim {
h[0] = Item{i, sim}
heap.Fix(&h, 0)
}
}
// ...
}延迟：1000 万次余弦计算 ≈ 100-500ms（多核并行后） QPS：单核约 2-10，需要几百台机器 精确性：100%
不可行，但作为 baseline 有价值（用来衡量近似算法的召回率）。
方案 B：近似最近邻（ANN）—— LSH
Locality Sensitive Hashing（局部敏感哈希）
核心思想：用一组哈希函数，让”相似的点以高概率哈希到同一个桶”。
对于余弦相似度，用随机超平面哈希：
type LSH struct {
numTables int // 哈希表的数量（提高召回率）
numBits int // 每个哈希的位数
hyperplanes [][]float32 // 随机超平面
tables []map[uint64][]int // 桶 → 用户 ID 列表
vectors [][]float32
}
func NewLSH(vectors [][]float32, numTables, numBits int) *LSH {
dim := len(vectors[0])
lsh := &LSH{
numTables: numTables,
numBits: numBits,
hyperplanes: make([][]float32, numTables*numBits),
tables: make([]map[uint64][]int, numTables),
vectors: vectors,
}
for i := range lsh.tables {
lsh.tables[i] = make(map[uint64][]int)
}
// 生成随机超平面
r := rand.New(rand.NewSource(42))
for i := range lsh.hyperplanes {
hp := make([]float32, dim)
for j := range hp {
hp[j] = float32(r.NormFloat64()) // 高斯分布
}
lsh.hyperplanes[i] = hp
}
// 建索引
for userID, v := range vectors {
for t := 0; t < numTables; t++ {
hash := lsh.computeHash(v, t)
lsh.tables[t][hash] = append(lsh.tables[t][hash], userID)
}
}
return lsh
}
// 计算哈希：每个超平面产生 1 bit（向量在超平面的哪一侧）
func (l *LSH) computeHash(v []float32, tableIdx int) uint64 {
var hash uint64
for b := 0; b < l.numBits; b++ {
hp := l.hyperplanes[tableIdx*l.numBits+b]
dot := float32(0)
for i := range v {
dot += v[i] * hp[i]
}
if dot > 0 {
hash |= (1 << uint(b))
}
}
return hash
}
// 查询：只计算同一个桶里的候选
func (l *LSH) Query(query []float32, k int) []int {
candidates := make(map[int]bool)
for t := 0; t < l.numTables; t++ {
hash := l.computeHash(query, t)
for _, id := range l.tables[t][hash] {
candidates[id] = true
}
}
// 在候选集里精确计算
h := make(SimilarityHeap, 0, k)
for id := range candidates {
sim := cosine(query, l.vectors[id])
// ... 维护 Top-K 堆
}
return extractIDs(h)
}关键性质：
随机超平面哈希的性质：
P(h(a) == h(b)) = 1 - θ(a,b)/π
其中 θ 是向量夹角
★ 夹角越小（越相似），哈希碰撞概率越高
★ 这正是"局部敏感"的含义
参数调优：
numBits大 → 桶更多更小 → 精确但召回率低numBits小 → 桶更少更大 → 召回率高但候选多numTables多 → 召回率高（多个表取并集），但内存和查询时间翻倍
典型配置：numBits=16, numTables=20 → 召回率 ~90%，查询只算几万个候选（而不是 1000 万）
性能：
- 查询延迟：~5-20ms（候选集几千到几万）
- 内存：20 个哈希表 × 1000 万条目 ≈ 几 GB
- 召回率：85-95%
方案 C：HNSW（分层可导航小世界图）—— 业界主流
这是目前工业界最好的 ANN 算法（Faiss、hnswlib、Milvus 都用它）。
核心思想：构建一个多层图，每一层都是”小世界网络”（大部分节点度很小，但少数节点有长距离连接）。
第 2 层（最稀疏）: A ──────────── G ← 只有几个节点，用于快速定位
\ /
第 1 层: A ── C ── E ── G ── I ← 中等密度
\ / \ / \ /
第 0 层（最密）: A-B-C-D-E-F-G-H-I-J-K-... ← 全部节点，精确搜索
查询过程：
1. 从第 2 层的任意节点开始
2. 贪心走到最近的节点（只在这一层内走）
3. 下降到第 1 层，继续贪心
4. 下降到第 0 层，做更彻底的搜索（efSearch 控制搜索宽度）
5. 返回 Top-K
为什么快？
层数 = O(log n)
每层步数 = O(log n)（小世界网络的性质）
总复杂度 = O(log n)
★ 1000 万节点，只需要几百次距离计算！
性能（1000 万 × 128 维）：
- 查询延迟：1-3 ms
- 召回率@10：95-99%
- 内存：约 6-8 GB（原始向量 + 图结构）
- QPS（单节点）：3000-5000
1 万 QPS 需要 2-4 台机器（加上副本和余量，实际 6-10 台）
方案 D：乘积量化（PQ）—— 内存极端受限时
核心思想：把 128 维向量切成 8 段，每段 16 维。每段用 K-means 聚成 256 个簇中心。
原始向量: 128 维 × 4 字节 = 512 字节
量化后: 8 段 × 1 字节（簇 ID）= 8 字节
★ 压缩 64 倍！
查询：预先算好查询向量与所有簇中心的距离表，查询时只查表。
性能：
- 内存：1000 万 × 8 字节 = 80 MB ✓
- 查询延迟：~10ms
- 召回率：70-85%（有损压缩，精度下降）
方案选择
| 方案 | 延迟 | 召回率 | 内存 | 机器数 | 适用 |
|---|---|---|---|---|---|
| 暴力 | 100-500ms | 100% | 5GB | 几百台 | 只做 baseline |
| LSH | 5-20ms | 85-95% | 几 GB | 10-20 台 | 简单，易实现 |
| HNSW | 1-3ms | 95-99% | 6-8GB | 6-10 台 | ★ 推荐 |
| PQ | ~10ms | 70-85% | 80MB | 4-6 台 | 内存受限 |
我的选择：HNSW
理由：
- 延迟最低（1-3ms），满足”实时”要求
- 召回率高（95-99%），业务可接受
- 有成熟的 Go 实现（hnswlib 的 Go binding，或自己实现）
- 支持增量更新（新用户可以随时插入）
系统架构
┌─────────────────────────────────────────────────────┐
│ 查询层 │
│ 1 万 QPS → 负载均衡 → N 个查询节点 │
└──────────────────────┬──────────────────────────────┘
│
┌──────────────────────▼──────────────────────────────┐
│ HNSW 索引集群（分片） │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Shard 0 │ │Shard 1 │ │Shard 2 │ ... │
│ │(250万) │ │(250万) │ │(250万) │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ 每个分片独立建图，查询时并行查所有分片，合并结果 │
└──────────────────────┬──────────────────────────────┘
│
┌──────────────────────▼──────────────────────────────┐
│ 特征服务（生成用户向量） │
│ 离线：每天全量更新 │
│ 实时：用户行为触发增量更新 │
└─────────────────────────────────────────────────────┘
为什么不建一个大图而要用分片？
单个图的缺点：
· 内存装不下（1000 万 × 图结构 ≈ 20GB+）
· 无法水平扩展
· 更新时锁粒度大
分片的优点：
· 每个分片 250 万，内存可控
· 可水平扩展（加机器就加分片）
· 更新只影响单个分片
代价：
· 查询要查所有分片，然后合并（网络开销）
· 但合并很简单（每个分片返回 Top-10，从 N×10 里再取 Top-10）
五个容易忽略的工程问题
1. 更新与查询的一致性
新用户注册后，它的向量要进索引。HNSW 支持增量插入，但插入会影响图结构。
方案：
- 写入用读写锁（读多写少场景，锁开销可接受）
- 或者双缓冲：写新索引，切换时原子替换指针
- 或者接受”短暂不一致”：新用户 5 分钟后才可被推荐（离线批量更新）
2. 删除用户（GDPR 合规）
HNSW 的删除很麻烦——删掉节点会破坏图的连通性。
方案：
- 软删除：标记删除，查询时过滤（简单但浪费空间）
- 定期重建：每天全量重建索引（推荐）
- 墓碑 + 重建：标记删除，累积到一定比例后重建
3. 向量维度与距离度量
余弦相似度 = 归一化后的内积。所以可以预先归一化所有向量，然后用内积代替余弦：
// 预处理：归一化
func normalize(v []float32) {
var norm float32
for _, x := range v {
norm += x * x
}
norm = float32(math.Sqrt(float64(norm)))
for i := range v {
v[i] /= norm
}
}
// 查询时：余弦 = 内积（因为已经归一化）
func dot(a, b []float32) float32 {
var sum float32
for i := range a {
sum += a[i] * b[i]
}
return sum
}这省掉了每次查询的归一化计算，快 30%。
4. SIMD 加速
距离计算是热点，用 SIMD 能快 4-8 倍：
// Go 里可以用 AVX2 汇编，或者用现成的库
// 比如 github.com/klauspost/cpuid 检测，然后调用优化的实现
// 或者用 Go 1.20+ 的试验性 SIMD 包
// golang.org/x/sys/cpu 检测 CPU 特性5. 召回率监控
这是最重要的，却最常被忽略。
近似算法会”悄悄”退化。你需要：
- 定期用暴力算法算一批查询的”真实 Top-10”
- 对比 HNSW 的结果，计算召回率
- 告警阈值设 90%
// 每天抽样 1000 个查询，用暴力算法验证
func monitorRecall(index *HNSW, vectors [][]float32, samples int) float64 {
totalRecall := 0.0
for i := 0; i < samples; i++ {
q := randomQuery()
approx := index.Query(q, 10)
exact := bruteForceTopK(q, vectors, 10)
// 计算召回率
hit := 0
exactSet := toSet(exact)
for _, id := range approx {
if exactSet[id] {
hit++
}
}
totalRecall += float64(hit) / 10.0
}
return totalRecall / float64(samples)
}为什么重要：数据分布会漂移。半年前调好的参数，现在可能召回率只有 70%。没有监控，你永远不会知道推荐质量在下降。
一句话总结
这个问题的核心不是”选哪个算法”，而是”理解你的约束”：
| 约束 | 影响 |
|---|---|
| QPS 1 万 | 暴力不可行，必须 O(log n) |
| 实时（< 10ms） | 排除外部排序、批处理 |
| 1000 万 × 128 维 | 内存 5GB，需要分片或压缩 |
| 推荐场景 | 允许 5% 的误差，不需要 100% 精确 |
当约束清楚了，算法选择几乎是唯一的。
这就是为什么”理解问题”比”记住算法”重要。 你不需要背下 HNSW 的细节，但你需要知道：
- 暴力不可行（算一遍就知道）
- 有 ANN 这个方向（知识面）
- HNSW 是目前最好的（调研能力）
- 要监控召回率（工程经验）
小结：这一节你应该带走的东西
-
七种范式各有适用场景：分治（可分解+可合并）、贪心（需证明）、DP（最优子结构+重叠子问题）、回溯（搜索+剪枝）、随机化（消除可预测性）、近似（NP-难问题）、流式/外部（内存受限）。
-
贪心的正确性需要证明，不能想当然。标准方法是”交换论证”。大部分”看起来能贪心”的问题是错的。
-
0-1 背包 vs 完全背包，只差一个遍历方向：逆序保证每个物品只选一次。
-
分支限界的关键是”乐观下界”：下界必须一定小于真实最优，否则会剪掉正确答案。
-
随机化最重要的价值是消除可预测性——防止算法复杂度攻击。Go 的 map 用随机哈希种子就是这个原因。
-
近似算法用”精确性”换”速度/空间”：Count-Min Sketch 用几 KB 空间处理 100GB 数据，代价是只高估不低估。
-
算法选择取决于约束，没有脱离约束的”最好算法”。推荐系统的案例说明了：算清楚账，约束清楚了，选择几乎是唯一的。
下一节
老陈的预告：为什么 MySQL 用 B+Tree 而不用红黑树？为什么 Redis 用跳表而不用 B+Tree？为什么 LevelDB 用 LSM-Tree？
答案不是”某个更好”，而是”它们各自针对不同的存储层次”。 下一节我们把数据结构和硬件特性对应起来，然后用 Go 从零写一个 B+Tree。