最近整理代码仓库的时候翻出了一套我自己一直在用的中等难度编程练习题。说是练习题其实更像是一套能力体检单每一道题都对应一个基本功方向做完一遍就能看出自己在并发、异步、网络、文本处理、递归回溯这几个常见领域里哪些是真正掌握的哪些只是以为自己会。这套题我前前后后给不少准备跳槽的朋友和团队新人做过反馈比较一致——难度不低但又不至于劝退认真做完一遍能感觉到自己的进步。如果你刚刷完入门题或者平时写业务代码挺顺手但一遇到底层机制就心里发虚这套卷子就是给你准备的。语言方面参考实现以 Python 为主个别地方我会提一下 C 的对应写法毕竟这两门语言在面试和实际工作里出现频率最高。下面直接进入正题。1. 中等卷到底在考什么1.1 什么叫做“中等难度”很多人对中等难度有误解以为就是算法题从 easy 换成 medium。我的判断标准不太一样中等难度的题往往是那种“单独看每个知识点你都会但凑到一起就容易翻车”的题。拿生产者-消费者模型举例。每个新手都知道线程和队列但要把容量控制、等待唤醒、退出条件、异常处理全部写对立刻就能筛掉一批自以为会并发的人。中等难度的核心不在于知识量而在于约束条件、边界情况、错误处理这些工程细节。你在刷这套卷子的时候重点不是“终于写出来了”而是“为什么我的第一版总是差点意思”。抓住这种感觉能力的提升就藏在里面。1.2 这套卷子怎么设计我挑题的原则只有一条覆盖日常开发里最容易踩坑的方向。算法数据结构当然重要但工作中真正折磨人的往往是并发、网络、文件处理这些基础能力。所以这套卷子分成了五个方向题目编程方向核心考点建议用时生产者-消费者模型并发编程锁、条件变量、线程安全60分钟异步批量超时控制异步编程事件循环、协程取消45分钟多客户端聊天室网络编程socket、广播、协议设计90分钟日志跨行解析与统计文本处理状态机、I/O 性能60分钟组合总和回溯算法递归、剪枝45分钟五个方向基本对应了后端开发、脚本自动化、底层调试里最高频的几类问题。做题的时候建议严格计时不要边写边看资料把每一次练习都当笔试来对待。做完再对照参考实现找出差异和盲区。2. 题目一用并发队列实现生产者-消费者模型2.1 题目描述与考察点写一个固定容量的阻塞队列支持put、get、size三个方法要求多线程环境下线程安全。然后基于这个队列写一个生产者-消费者程序若干生产者线程随机产生任务并放入队列若干消费者线程从队列取出任务并模拟处理最后统计成功处理的总任务数。这道题考察的是并发编程最内核的东西互斥、同步、等待与唤醒。很多人能直接调queue.Queue把题目糊弄过去但那不行这里必须让你自己实现队列内部的保护逻辑否则就失去了练习的意义。queue.Queue当然可以用但那是验答案的参照物不是你的实现。2.2 参考实现Python threading 版import threading import random import time from collections import deque class BlockingQueue: def __init__(self, capacity): self.capacity capacity self.queue deque() self.lock threading.Lock() self.not_full threading.Condition(self.lock) self.not_empty threading.Condition(self.lock) def put(self, item): with self.not_full: while len(self.queue) self.capacity: self.not_full.wait(timeout5) self.queue.append(item) self.not_empty.notify() def get(self): with self.not_empty: while not self.queue: self.not_empty.wait(timeout5) item self.queue.popleft() self.not_full.notify() return item def size(self): with self.lock: return len(self.queue)生产者消费者主流程可以这样写def producer(q, count, result): for i in range(count): q.put(i) time.sleep(random.random() * 0.01) result.put(done) def consumer(q, result): total 0 while True: item q.get() if item is None: # 哨兵值退出 break total item time.sleep(random.random() * 0.02) result.put(total)这里我用了哨兵值None来通知消费者退出这是并发编程里非常常见的技巧。主线程把所有任务发完后往队列里放入 n 个None和消费者数量相同每个消费者拿到None就退出。这么做能避免直接terminate线程带来的数据丢失和资源泄漏问题。2.3 容易翻车的并发细节我在让别人做这道题的时候见过太多了第一wait必须放在循环里不能简单用if。官方文档明确说存在虚假唤醒的可能而且就算没有虚假唤醒多个消费者同时被唤醒后队列也可能已经被另一个线程取空了。用while重新检查条件是标准做法。第二两个Condition共用一个Lock是合理的。not_full和not_empty都基于同一个self.lock这保证了put和get内部的互斥是同一把锁不会出现一个线程在put、另一个线程在get时同时修改队列结构的竞争问题。第三wait(timeout5)这个 timeout 很关键。如果某个消费者线程在get时恰好遇到全局异常或者生产者全部崩溃其他消费者会被一直卡住。加上 timeout 可以保证队列在极端情况下不会无限期挂死虽然我平时写生产环境代码时也会加但很多人第一次根本想不到。第四千万别在想当然的地方手动释放锁。初学者经常在wait之后尝试lock.release()结果把Condition的内部锁逻辑完全打乱。只要用了with self.not_full:这种写法锁的获取和释放都交给上下文管理器你只需要在条件满足时调用notify()。3. 题目二异步任务批量执行与整体超时控制3.1 场景与题目描述现实里这个需求太常见了批量调用第三方接口、批量抓网页、批量查数据库每一个请求都可能很慢但整体耗时又不能无限拖。这道题要求写一个通用函数输入一批协程对象设置总超时时间返回已完成任务的结果列表超时未完成的任务全部取消。用多线程也能做但协程在大量 I/O 等待场景下资源占用低得多。Python 的asyncio是主要考察目标如果你是 C 方向可以考虑用std::future配合wait_for来做思路是一样的。3.2 asyncio 参考实现import asyncio async def run_wait_timeout(tasks, timeout): tasks [asyncio.ensure_future(t) for t in tasks] done, pending await asyncio.wait(tasks, timeouttimeout) for p in pending: p.cancel() # 等取消真正完成避免任务泄漏 if pending: await asyncio.gather(*pending, return_exceptionsTrue) results [] for d in done: try: results.append(d.result()) except Exception as e: results.append(ferror: {e}) return results调用示例async def demo(): async def fake_request(i, delay): await asyncio.sleep(delay) return i tasks [fake_request(i, 0.5 if i % 2 0 else 2.0) for i in range(5)] results await run_wait_timeout(tasks, timeout1.0) print(results)asyncio.wait默认是等待所有任务完成传入timeout参数后超时时会返回两个集合done表示已完成pending表示超时未完成。接下来要做的就是把pending里的任务逐个取消再用gather等一下取消动作真正完成。3.3 几个必须注意的坑第一asyncio.ensure_future之后任务会立刻进入事件循环的调度队列。如果你在run_wait_timeout之外单独创建协程但从不 await控制台会打出一堆“coroutine was never awaited”的警告严重的还会造成协程泄漏。所以函数内部一定要把任务对象收集起来统一管理。第二cancel()不是同步生效的。调用cancel只是给任务发了一个取消请求如果任务正处于某个await点上需要事件循环再跑一圈才能真正进入取消流程。所以取消后必须await asyncio.gather(*pending, return_exceptionsTrue)否则程序退出的时候可能看到 “Task was destroyed but it is pending” 的报错。第三子协程内部不要瞎捕获CancelledError。有些人为了“不让程序报错”在业务协程里写except Exception结果把asyncio.CancelledError也吞掉了取消请求就变成了一张空头支票。正确做法是如果确实需要在协程退出前做一些清理捕获之后重新raise把取消动作继续传递下去。第四如果有任务在超时瞬间刚好完成它的结果会进done而不是pending这是不可控的边界行为。如果你的业务要求“超过一秒就必须视作失败”那就要自己在协程外层再包一层asyncio.wait_for给每个任务单独加超时时间而不是只靠总超时。4. 题目三多客户端 TCP 聊天室4.1 题目描述用 Socket 写一个局域网聊天室要求支持多个客户端同时连接每个客户端发送的消息其他所有客户端都能收到客户端有昵称进入和退出时广播提示。这个题目的工程味很浓考察点也不只是socket本身还包括线程模型、资源清理、部分客户端异常断开时如何保证服务端不崩溃。4.2 服务端实现要点最直观的方案是threading加socket每个客户端连接开一个线程处理主线程循环等待新连接。核心服务端逻辑可以这样写import socket import threading clients [] clients_lock threading.Lock() def broadcast(message, senderNone): with clients_lock: for conn in clients[:]: if conn is sender: continue try: conn.sendall(message) except OSError: # 发送失败说明客户端已经断开清理掉 clients.remove(conn) def handle_client(conn, addr): try: while True: data conn.recv(1024) if not data: break broadcast(data, senderconn) except OSError: pass finally: with clients_lock: if conn in clients: clients.remove(conn) conn.close() def start_server(host0.0.0.0, port9000): server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen(5) while True: conn, addr server.accept() with clients_lock: clients.append(conn) threading.Thread(targethandle_client, args(conn, addr), daemonTrue).start()几个关键点全局clients列表是多个线程共享的任何修改都要在clients_lock保护下进行否则broadcast遍历时另一个线程刚好remove就会抛RuntimeError。我习惯遍历的时候用clients[:]做一次快照这样即使在遍历过程中有连接断开也只会影响快照本身不会造成列表结构被修改的异常。4.3 协议设计、粘包与断连检测这道题最容易忽略的是粘包问题。TCP 是流式协议它不保证一次send对应一次recv多条消息可能被合并在一起到达也可能一条消息被拆成多次到达。最省事的办法是用换行符分隔消息约定每条完整消息以\n结尾。客户端sendall时在末尾手动加\n服务端recv后按缓冲区里的\n切分攒够一整行再广播。如果题目要求更严谨可以用“长度前缀”的方式先发 4 字节表示消息长度再发正文但那个实现复杂一些做题阶段用换行符完全够用。断连检测也有门道。recv返回空字节串b表示对端正常关闭但如果是客户端直接拔网线或断电服务端可能在很长一段时间内读不到任何数据也收不到断连通知。我见过不少人在这里卡住以为掉线的客户端永远凑不够数据线程就一直挂着。实际练习里可以再加一个socket.settimeout(30)超时后抛出socket.timeout这时候就当成断连处理把连接清理掉。当然这个超时时间要根据业务场景调整聊天室 30 秒没动静就断开可能有点激进但练习阶段用来理解超时机制是很有用的。客户端的实现相对简单但要分两个线程一个负责读标准输入并发送一个负责接收服务端广播并打印。很多新手只写了一个线程结果recv阻塞住输入也无法发送最后不得不 CtrlC 退出。5. 题目四日志文件跨行解析与统计5.1 题目描述写一个命令行工具解析一个日志文件日志中每条消息可能跨越多行每条消息以[MSG]标记开头消息内部可能包含错误码字段格式类似[ERR500]。要求统计每个错误码出现的次数输出 Top N按次数降序排列。这个题目在工作中很实用因为真实日志往往不是一行一条的。异常堆栈、多行 SQL、JSON 格式化后的日志都会把一条完整消息撑成好几行。如果你只会用readlines配合正则硬匹配很容易把一条消息拆成多段统计结果一团糟。5.2 状态机实现核心思路是维护一个状态变量当前是否处在一条消息内部。遇到[MSG]则进入“消息内”状态开始累积内容非[MSG]行并且处于“消息内”状态则继续累积遇到下一个[MSG]或文件结束就把之前累积的内容作为一条完整消息处理。def extract_error_code(block): # block 是完整消息的行列表 for line in block: if [ERR in line: start line.index([ERR) len([ERR) end line.index(], start) return line[start:end] return unknown def parse_log(path, top_n10): stats {} current_block [] in_message False with open(path, r, encodingutf-8) as f: for line in f: line line.rstrip(\n) if line.startswith([MSG]): if in_message and current_block: code extract_error_code(current_block) stats[code] stats.get(code, 0) 1 current_block [line] in_message True elif in_message: current_block.append(line) # EOF 处理最后一条 if in_message and current_block: code extract_error_code(current_block) stats[code] stats.get(code, 0) 1 return sorted(stats.items(), keylambda x: x[1], reverseTrue)[:top_n]5.3 为什么不用一个正则搞定我知道有人看到这个题目第一反应是“我可以用正则匹配整个文件。”比如re.findall(r\[MSG\].*?(?\[MSG\]|$), text, re.S)。能用但对大文件非常不友好你需要把整个文件读进内存.*?配合re.S在极端情况下还会导致灾难性回溯一个几百 MB 的日志能把 CPU 吃满。状态机的好处在于它是流式的逐行处理内存占用只和单条日志的最大长度有关而不是整个文件大小。这是一个非常重要的工程思维能边读边处理的数据就不要全量加载。另外要注意extract_error_code的实现。理想情况下错误码字段固定写在某一行但不排除它出现在多行拼接后的中间位置所以我选择遍历整个 block 逐行查找。当然这也有陷阱如果错误码本身被换行拆断比如一行是[ERR50下一行是0]上面的写法就会漏掉。真实场景里这个要看日志格式约定如果格式不可靠就得在current_block累积完之后把内容拼成一个长字符串再做正则。5.4 性能与边界日志文件解析的边界问题很典型空文件、文件只有一条消息但末尾没有换行、错误码完全缺失。这些情况在写代码时都要想到。我建议的检查顺序是先跑一个空文件确认程序不报错再跑一个正常样例确认统计数量最后故意删掉最后一条消息的换行符确认 EOF 分支能兜住最后一条日志。从性能角度讲逐行for line in f已经是很好的做法了。不要用f.read()也不要用f.readlines()前者把整个文件塞进内存后者会创建一个存有所有行的列表。一行一行迭代时Python 内部有自己的缓冲区读取效率并不低。6. 题目五组合总和问题的回溯与剪枝6.1 题目描述给定一个无重复元素的数组candidates和一个目标数target找出所有可以使数字和为target的组合。candidates中的数字可以无限重复使用。要求返回所有不重复的组合按任意顺序。这是回溯算法的经典代表笔试里经常以各种面目出现。它考察你对递归、状态恢复、剪枝的理解。很多人一看到“可以无限重复使用”就懵了不知道递归的下一层该从哪里开始。实际上这个条件恰恰是解题的关键。6.2 回溯实现def combination_sum(candidates, target): candidates.sort() res [] path [] def dfs(start, remain): if remain 0: res.append(path[:]) return for i in range(start, len(candidates)): if candidates[i] remain: break path.append(candidates[i]) dfs(i, remain - candidates[i]) path.pop() dfs(0, target) return res核心逻辑在dfs(i, remain - candidates[i])因为每个数字可以无限重复使用所以递归下一层仍然从当前下标i开始。这样就能生成[2, 2, 3]这种重复选同一数字的组合。同时因为是按数组顺序向后扫描[2, 3, 2]和[3, 2, 2]永远不可能出现从根上避免了重复。6.3 剪枝细节剪枝是这道题的灵魂。最直观的剪枝是排序后判断candidates[i] remain直接break。因为数组已经升序排列后面的数只会更大继续循环没有意义。这个剪枝能把很多无用的递归分支直接砍掉尤其是在 target 较小、数组较大的时候效果非常明显。还有一个容易被忽略的细节res.append(path[:])必须用切片复制一份path而不是直接res.append(path)。如果直接追加引用回溯回去执行path.pop()时已保存的结果也会被一起改掉最后res里全是空列表。这个知识点我称之为“回溯的一滴血”所有初学者都会在这上面挂一次。如果题目稍作变化变成“每个数字只能使用一次”那递归下一层就要改成dfs(i 1, remain - candidates[i])并且还要处理数组中有重复数字导致结果重复的问题通常需要加一个if i start and candidates[i] candidates[i - 1]: continue的跳过逻辑。做题时建议把这两个变体都写一遍对比差异理解会更扎实。7. 整套卷子的避坑清单7.1 高频错误速查表我把这些年带人做这套题遇到的高频错误整理成一张表每一条都是我实际见过、也踩过的题目方向高频错误排查思路生产者-消费者死锁检查锁顺序是否一致、是否在持锁时递归调用生产者-消费者忙等用条件变量和wait不要用sleep空转生产者-消费者消费者退出不了使用哨兵值或显式退出标记异步超时任务泄漏取消之后必须await gather异步超时异常被静默吞掉在done集合里逐个调d.result()取异常聊天室粘包/半包用换行符或长度前缀做消息边界聊天室send 抛异常导致线程退出捕获OSError并清理连接日志解析EOF 丢最后一条循环结束后补一次状态刷新日志解析内存占用高逐行读不要read()组合总和结果重复start从当前索引开始不要每次回 0组合总和保存结果被修改path[:]复制后再放入结果集7.2 调试方法我自己的调试习惯是并发和异步代码先跑最小样例把线程数降低到 2 个任务数降低到 3 个再加上关键位置日志。比如生产者每放一个任务打印一次消费者每取一个任务打印一次这样死锁还是活锁、数据丢没丢一眼就能看出来。第二个习惯是写测试。不要觉得练习题不需要测试这几道题都非常适合用单元测试固定正确行为。生产者-消费者可以断言最终统计总数等于生产总数异步超时可以分别测“全部按时完成”和“部分超时”两种场景聊天室可以直接在本地起服务端用两个客户端脚本验证广播日志解析可以手写一个 20 行的小样本来验证统计结果。这些测试用例本身就是你加深理解的一部分。第三永远不要相信“看起来是对的”。每次改动后都重新跑一遍所有样例哪怕只是改了一个变量名。我吃过太多亏改了一行以为不影响逻辑结果刚好破坏了某个边界条件。练习题也是代码要用生产代码的严谨性来对待。写在最后这套卷子我自己每年都会刷一遍每次都能发现新的盲区。第一次做生产者-消费者的时候我觉得这题太简单了写完之后一跑消费者全部卡死后来才意识到自己对条件变量的底层机制理解不够。后面几道题也差不多总是写着写着发现“原来这个地方我没理解透”。建议你做的时候也别急着看答案先自己写完跑起来再对照参考实现逐行检查。有差异说明这套题对你有价值完全没差异说明你的基本功比我强可以直接跳去刷更难的题目了。