算法

围绕“公开接口先行、helper 以契约占位、实现围绕不变量展开”这条主线，系统讲解如何从算法题过渡到中型程序设计，并用 LRUCache 与下单流程示例说明它和 DDD 的分工关系。

用一个可运行的 Go 版本基础消息代理，讲透发布订阅、重试语义、失败契约、吞吐与积压估算，以及从朴素实现到工程可用实现的关键取舍。

副标题 / 摘要 “相似商品推荐”的核心是共购关系。本文用最小协同过滤思路解释实现方法。 目标读者 需要搭建推荐功能的工程师 负责电商系统的开发者 学习基础推荐算法的人 背景 / 动机 推荐系统能提升转化率与停留时间。 最简单的实现方式是基于“共购/共点”统计。 核心概念 协同过滤：基于用户行为的相似性 共购矩阵：商品一起出现的次数 召回与排序：先找候选，再排序 实践指南 / 步骤 收集用户行为（购买/浏览） 统计共现关系 生成候选集 结合热度或规则排序 可运行示例 from collections import Counter def recommend(orders, item): co = Counter() for order in orders: if item in order: for x in order: if x != item: co[x] += 1 return [x for x, _ in co.most_common(3)] if __name__ == "__main__": orders = [ ["A", "B", "C"], ["A", "B"], ["A", "D"], ["B", "C"], ] print(recommend(orders, "A")) 解释与原理 协同过滤假设“经常一起出现的商品更相关”。 这是冷启动与小规模电商的常用起点。 ...

副标题 / 摘要 REPL 是交互式程序的最小形态。本文先构建 echo REPL，再扩展为逆波兰计算器。 目标读者 想练习解析与交互程序的开发者 学习表达式求值的人 初中级算法学习者 背景 / 动机 交互式解释器是语言与工具链的核心。 从简单 REPL 演化到计算器，是理解解析流程的好练习。 核心概念 REPL：读入-求值-输出循环 逆波兰表达式（RPN）：无需括号的表达式形式 栈求值：用栈完成运算 实践指南 / 步骤 先实现 echo REPL（读入并输出） 加入退出指令（如 quit） 解析输入为 token 列表 用栈计算 RPN 表达式 可运行示例 import sys def eval_rpn(tokens): stack = [] for t in tokens: if t in {"+", "-", "*", "/"}: b = stack.pop() a = stack.pop() if t == "+": stack.append(a + b) elif t == "-": stack.append(a - b) elif t == "*": stack.append(a * b) else: stack.append(a / b) else: stack.append(float(t)) return stack[-1] def repl(): while True: line = input("> ").strip() if line == "quit": return if not line: continue try: tokens = line.split() print(eval_rpn(tokens)) except Exception as e: print("error:", e) if __name__ == "__main__": repl() 解释与原理 RPN 的关键是“运算符后置”，因此可以用栈自然求值。 REPL 只需持续读入、求值、输出即可。 ...

副标题 / 摘要 内存无法容纳 10GB 数据时，外部排序是标准方案。本文介绍分块排序与多路归并。 目标读者 需要处理大文件的工程师 学习外部排序算法的读者 关注磁盘 I/O 优化的人 背景 / 动机 当数据量超过内存容量，传统内存排序会失败。 外部排序通过“分块 + 多路归并”解决问题。 核心概念 分块排序：把大文件拆成可放入内存的小块 多路归并：合并多个有序块 磁盘 I/O：顺序读写优于随机读写 实践指南 / 步骤 按内存容量分块读取 每块内存排序并写回磁盘 多路归并所有有序块 尽量顺序读写减少随机 I/O 可运行示例 # 简化示例：分块排序 + 归并 import heapq def merge_sorted(chunks): heap = [] for i, chunk in enumerate(chunks): if chunk: heapq.heappush(heap, (chunk[0], i, 0)) result = [] while heap: val, i, j = heapq.heappop(heap) result.append(val) if j + 1 < len(chunks[i]): heapq.heappush(heap, (chunks[i][j + 1], i, j + 1)) return result if __name__ == "__main__": chunks = [sorted([3, 1, 2]), sorted([9, 7, 8])] print(merge_sorted(chunks)) 解释与原理 外部排序的核心是把“大问题拆成小问题”，每块可在内存中排序。 最后通过多路归并生成整体有序序列。 ...

副标题 / 摘要 10TB 数据无法在单机完成排序。本文介绍分布式排序的核心流程与工程要点。 目标读者 处理大规模数据的工程师 学习分布式系统的开发者 关注数据处理流程的人 背景 / 动机 数据规模超过单机能力时，必须通过分布式拆分与并行归并完成排序。 这涉及数据切分、网络传输与容错。 核心概念 分片（Shard）：数据切分到多个节点 Shuffle：按 key 重新分配数据 分布式归并：跨节点合并有序块 实践指南 / 步骤 按范围或哈希切分数据 各节点本地排序 Shuffle 让相同范围的数据聚集 全局归并并输出结果 可运行示例 # 简化的“分片排序”示意 def shard_sort(chunks): return [sorted(c) for c in chunks] def merge_two(a, b): i = j = 0 res = [] while i < len(a) and j < len(b): if a[i] < b[j]: res.append(a[i]); i += 1 else: res.append(b[j]); j += 1 res.extend(a[i:]) res.extend(b[j:]) return res if __name__ == "__main__": shards = shard_sort([[3, 1], [4, 2]]) print(merge_two(shards[0], shards[1])) 解释与原理 分布式排序的关键在于“局部排序 + 全局归并”。 Shuffle 会成为主要瓶颈，需要优化网络与分区策略。 ...

副标题 / 摘要 碎片整理的目标是减少随机寻道，提高顺序读写性能。本文给出设计流程与简化实现。 目标读者 学习系统设计的开发者 关注存储性能的工程师 想理解“数据布局优化”的人 背景 / 动机 文件频繁增删会导致数据块分散。 碎片整理通过“重排布局”提升连续读写效率。 核心概念 碎片：文件块分散在多个位置 压缩（Compaction）：把数据块向前聚集 元数据更新：移动块后更新索引 实践指南 / 步骤 扫描磁盘找到空洞与数据块 规划移动顺序，避免覆盖 逐块移动并更新元数据 验证一致性并生成报告 可运行示例 # 简化模型：1 表示数据块，0 表示空洞 def defragment(blocks): write = 0 for read in range(len(blocks)): if blocks[read] == 1: blocks[write], blocks[read] = blocks[read], blocks[write] write += 1 return blocks if __name__ == "__main__": data = [1, 0, 1, 0, 1, 0, 0, 1] print(defragment(data)) 解释与原理 通过双指针把所有数据块向前移动，实现“紧凑布局”。 现实文件系统还需要处理文件连续性与元数据同步。 常见问题与注意事项 整理期间如何保证数据不丢？ 需要日志与校验。 整理会影响系统性能吗？ 会，通常在低峰期进行。 SSD 还需要碎片整理吗？ 需求较低，但仍需维护磨损均衡。 最佳实践与建议 使用快照或日志保护数据 控制整理窗口，避免影响业务 定期评估碎片率 小结 / 结论 碎片整理是“数据布局优化”的工程问题，需要性能与安全的平衡。 核心在于安全移动与元数据一致性。 ...

副标题 / 摘要 随机数生成是很多算法的基础。本文用线性同余法实现一个可复现的 rnd()。 目标读者 学习随机算法的开发者 需要理解 PRNG 的工程师 算法与系统基础学习者 背景 / 动机 大多数语言的随机数来自伪随机算法。 理解基础实现有助于评估随机性与可复现性。 核心概念 伪随机数（PRNG）：由公式生成的序列 种子（Seed）：决定序列起点 可复现性：同样种子产生相同序列 实践指南 / 步骤 选择线性同余参数 用种子初始化状态 每次调用更新状态并输出 将结果归一化到 [0,1) 可运行示例 class LCG: def __init__(self, seed=1): self.mod = 2 ** 31 self.a = 1103515245 self.c = 12345 self.state = seed def rnd(self): self.state = (self.a * self.state + self.c) % self.mod return self.state / self.mod if __name__ == "__main__": rng = LCG(seed=42) for _ in range(3): print(rng.rnd()) 解释与原理 线性同余法是最简单的 PRNG： state = (a * state + c) mod m。 它速度快，但随机性质量有限。 ...

副标题 / 摘要 标记-清除是最基础的垃圾回收模型。本文用简化示例解释“可达性”与回收过程。 目标读者 想理解 GC 原理的开发者 系统编程与语言设计学习者 关注内存管理的工程师 背景 / 动机 手动内存管理容易出错，而 GC 通过“可达性”自动回收对象。 理解基础模型有助于调试内存问题。 核心概念 根集合（Roots）：直接可达的对象 可达性：从根出发可访问到的对象 标记-清除：标记存活对象，清除不可达对象 实践指南 / 步骤 构建对象图与引用关系 从根集合进行标记遍历 清除未被标记的对象 输出回收结果 可运行示例 class Obj: def __init__(self, name): self.name = name self.refs = [] self.marked = False def mark(obj): if obj.marked: return obj.marked = True for r in obj.refs: mark(r) def sweep(heap): return [o for o in heap if o.marked] if __name__ == "__main__": a = Obj("a") b = Obj("b") c = Obj("c") a.refs.append(b) heap = [a, b, c] roots = [a] for r in roots: mark(r) heap = sweep(heap) print([o.name for o in heap]) # c 被回收 解释与原理 GC 的核心是假设“不可达对象可以回收”。 标记阶段找到存活对象，清除阶段释放其他对象。 ...

副标题 / 摘要 随机迷宫生成常用深度优先回溯法。本文解释思路并提供可运行实现。 目标读者 学习图算法的开发者 想做程序化生成的工程师 算法入门学习者 背景 / 动机 迷宫生成是图遍历与随机化的经典结合。 它能帮助理解 DFS、回溯与边界处理。 核心概念 网格图：迷宫格点和通道 DFS 回溯：随机探索与回退 墙与通路：用字符表示结构 实践指南 / 步骤 初始化全墙网格 从起点开始 DFS 随机选择未访问邻居并打通墙 回溯直到全部访问完成 可运行示例 import random def maze(w, h): grid = [["#"] * (2 * w + 1) for _ in range(2 * h + 1)] visited = [[False] * w for _ in range(h)] def carve(x, y): visited[y][x] = True dirs = [(1, 0), (-1, 0), (0, 1), (0, -1)] random.shuffle(dirs) for dx, dy in dirs: nx, ny = x + dx, y + dy if 0 <= nx < w and 0 <= ny < h and not visited[ny][nx]: grid[y * 2 + 1 + dy][x * 2 + 1 + dx] = " " grid[y * 2 + 1][x * 2 + 1] = " " grid[ny * 2 + 1][nx * 2 + 1] = " " carve(nx, ny) carve(0, 0) return "\n".join("".join(row) for row in grid) if __name__ == "__main__": print(maze(6, 4)) 解释与原理 DFS 回溯保证每个格子被访问一次，随机方向带来多样性。 打通墙壁即可形成迷宫通路。 ...