Python进阶与并发编程07：CPU工作为什么要换进程，以及为什么换完可能更慢-北京尧图网络

Python进阶与并发编程07CPU工作为什么要换进程以及为什么换完可能更慢报价系统不一定只等待网络。假设我们还要给每个SKU生成一份校验摘要算法包含大量Python整数运算。这个计算与访问供应商相互独立可以按商品分开执行。上一课的线程池适合把HTTP等待重叠但在常见启用GIL的CPython中它通常不能让多个线程同时执行这段纯Python循环。本课先构造可重复、无网络的CPU工作再用进程池执行观察摘要、进程ID和总耗时。算法只是学习用的滚动校验值不是密码学哈希也不能用于密码、数字签名或防篡改。选择它的原因很具体主要运算发生在Python层输入很小输出很小便于看清并行计算与进程启动两类成本。E06答案限制两个线程不限制结果的确定性E06要求三个供应商最多占用2个工作线程并把结果按总价、供应商名称稳定排序。answer_previous.py先创建一个共享同步Client再创建ThreadPoolExecutor(max_workers2)。三个Future交给as_completed收集退出线程池后再退出客户端确保还在运行的线程不会访问已关闭的连接池。答案同时用锁保护活动计数确认峰值不超过2、结束后为0并输出peak_threads至少两个任务真实重叠由第06课Barrier测试确定性验证。正常购买2件的结果必须是boreal 25900分、atlas 26400分cedar库存1件被过滤。答案另外构造两个相同总价的报价行断言atlas排在boreal之前。这一部分不能只用正常数据验证因为正常数据没有触发并列价格的第二排序键。上一课的活动计数测试进一步证明同时运行的线程恰好能达到2、又不会超过2避免“碰巧输出正确但其实串行”的实现漏网。进程边界改变了数据的旅行方式线程共享同一进程的地址空间传一个普通对象通常只是把引用交给另一个线程进程有各自独立的内存空间父进程不能把自己的对象地址当作子进程可用的地址。ProcessPoolExecutor会把函数身份、参数和结果通过序列化与进程间通道传递。本课传入(index, sku, rounds)三个简单值返回含序号、SKU、摘要和PID的字典。不要把HTTP客户端、打开的文件、线程锁或连接池塞进参数。它们要么不能pickle要么即使某种操作系统碰巧复制了底层资源也没有形成安全的跨进程使用合同。网络报价留在HTTP并发模型里CPU计算留在进程工作者中接口只穿过真正需要的数据。checksum放在模块顶层这是让spawn子进程能够通过模块名找到函数。嵌套函数和lambda没有一个可稳定重新导入的顶层名字常见结果是pickle错误。if __name__ __main__:保护的不是计算函数定义而是启动进程池的动作导入模块时应允许定义函数不应再次启动一批新进程。从这里可以推导出一个实用规则模块顶层放定义主入口里放启动。数据库连接、线程池和进程池不要在导入时偷偷创建。这个规则不仅避免Windows的递归启动也让测试、命令行和以后Web服务的生命周期更加清晰。显式spawn让Windows和Linux学到同一件事本课使用multiprocessing.get_context(spawn)再传给执行器的mp_context。spawn启动新的解释器重新导入所需模块因此启动成本容易观察也要求我们把初始化写正确。不同Python版本、平台的默认启动方法可能不同课程显式选择能避免在一台机器上能跑、换到另一台才暴露导入问题。freeze_support()放在主入口保护内部便于将来使用某些打包后的Windows可执行程序普通脚本下它通常没有额外动作。它不能代替主入口保护不能让lambda突然变成可pickle的函数也不消除子进程创建成本。不要把整段进程池演示复制到交互式控制台就期望同样行为。某些交互环境没有可被子进程导入的主模块。将文件保存到磁盘再从终端执行python lessons/07/main.py才是本课明确支持的运行方式。结果先证明一致再看速度程序先串行算四个SKU再用两个进程工作者算相同数据。它用索引把进程结果排序回来然后逐项对比摘要。PID只用于观察执行位置不参与摘要结果。若摘要不一致先修复数据处理不能拿错误输出的快速程序参加性能比较。as_completed仍按完成情况交付所以输出字段叫completion_order。不能用SKU当字典键恢复顺序因为真实批次可能包含重复SKU后一个会覆盖前一个。索引才是每个输入位置的唯一身份。这个经验将在批量报价接口里复用业务内容相同不代表请求位置相同。从课程根目录执行CPU实验不需要供应商服务只有E06答案需要先启动provider_lab.py。Linux将python替换为python3.11。python lessons/07/main.py python lessons/07/main.py--rounds 800000 python lessons/07/answer_previous.py python-m unittest discover-s lessons/07-p test_lesson.py-vrounds决定每项CPU工作量。2026-10-09本课程制作环境的一次实际运行中默认80000轮的串行阶段为168.19ms进程阶段为1885.08ms。这个数字只描述该次Windows/Python3.11.9运行它已经包含新建进程池、启动spawn工作者、传入任务、等待结果及关闭池并不是纯计算核耗时。其他机器和后台负载会不同。输出还可能显示四项任务都由同一个PID完成。max_workers2表示允许至多两个工作者不保证短任务一定公平分散如果一个进程先准备好另一个还在启动它可能已经把任务做完。不能写一个测试强制要求“小任务必须出现两个PID”那是在把偶然调度当成业务合同。用一个成本式决定是否值得并行串行耗时可以近似看成各项计算时间相加。进程方案还增加启动、输入序列化、传输、输出反序列化、结果收集等开销只有可并行计算足够大重叠节省的时间才可能覆盖这些成本。把一个很小函数交给进程执行往往只是让它多坐了一趟昂贵的车。加大任务粒度是一种办法例如让每个任务处理一小批SKU减少提交次数但批次越大也越难负载均衡。长期服务还可以复用进程池把启动成本摊到多次请求与此同时必须安排应用关闭时的收尾并限制待处理任务数量。课程本节每次新建池是为了展示完整成本不代表生产服务每个请求都应新建池。worker数量也不是越多越好。进程各自持有解释器和模块内存数量增加会放大内存、调度和CPU竞争。若Web服务已经启动四个worker每个worker再开四个进程池成员就可能出现十六个计算进程。容量预算必须看整台机器以及容器的实际CPU配额而不只看某个局部max_workers值。异常与关闭不因换成进程而消失普通Python异常会通过Future传回父进程调用future.result()时重新抛出。工作进程突然退出则可能让池进入损坏状态出现BrokenProcessPool这和某一个输入校验失败不是同一种错误。输入非法应在明确的数据边界识别池损坏则需要更高层决定如何恢复。像线程池一样对Future等待超时并不等于成功终止正在运行的计算。若任务有外部副作用简单重试还可能重复执行。本文摘要函数无副作用易于对比和重算这只是本实验性质不能推广成“所有进程任务失败都能自动重试”。测试会实际启动spawn进程并比较每项结果不依赖HTTP模拟。其目的在于证明文件入口、序列化和函数可导入确实成立同时确认没有漏任务。运行测试比单纯compileall更有意义代码即使能编译也可能在子进程重新导入时出错。E07每项都有摘要但输出必须对应原位置原题E07输入[KB-003, KB-001, KB-002]使用spawn进程池和顶层checksum函数生成每项校验摘要返回包含index, sku, checksum, pid的行。无论任务完成顺序如何返回行的index必须为[0,1,2]SKU顺序与原输入完全一致且每项摘要与相同轮数的串行计算一致。不能用SKU作唯一字典键。下一课给完整答案。实现时先预留与输入等长的列表把Future映射到原索引再在完成时写回对应位置。父进程拥有结果列表不要尝试让子进程直接修改父进程的普通列表。普通内存没有自动跨过进程边界明确传回数据反而更容易检验。父进程里的变量不会自动变成共享变量在子进程中向普通列表追加一个结果父进程的同名列表不会随之出现新元素。即使两个变量有相同名字它们也属于不同解释器的内存。某些启动方式会让子进程初始状态看起来像父进程但后续写入仍不等于共享修改。让计算函数显式返回结果再由父进程汇总比依赖隐含共享容易推理。确实需要跨进程共享时可以研究共享内存、进程安全队列或代理对象不过每种方式都带来同步、生命周期和序列化成本。本课不引入它们是因为摘要问题天然适合纯函数映射输入完整描述工作输出完整描述结果。先识别问题本身是否可拆比先选择最复杂的并发工具更关键。摘要中的索引也不要由工作进程的完成计数生成。不同进程各自从零开始计数会产生重复序号索引应由父进程读取输入时确定随参数传进去随结果传回来。PID是执行位置索引是业务位置这两类标识承担不同责任。小输出不代表输入搬运免费如果以后把商品图片或一个巨大的价格表随每项任务传入序列化可能比摘要运算还贵。重复传同一个大表尤其容易造成父进程忙着打包、子进程等着数据的情况。此时可以考虑在工作者初始化时加载只读数据或让任务只携带数据位置和必要参数但必须明确文件版本、可读性与清理责任。反过来工作者返回一个巨大的中间结构也会把压力搬回父进程。并行化设计应同时审视输入和输出而不只看计算循环。当前练习刻意让参数是短字符串和整数结果只是几个标量帮助我们先观察启动开销以后优化真实业务时必须重新测量数据移动成本不能沿用这个小实验的结论。最后还有不可并行的部分读取配置、准备输入、汇总排序和发送响应仍可能在主进程中串行运行。即使计算部分可以理想地分给很多核心总时长也受这些串行阶段限制。发现加进程后收益变小先按阶段计时而不是立即认为执行器实现有问题。本课完整源码main.py是独立CPU实验answer_previous.py和随课网关、规则副本只用于完成E06。它们不共享进程池也不把网络客户端传给子进程。lessons/07/answer_previous.pyE06: two worker threads and deterministic business ranking.fromconcurrent.futuresimportThreadPoolExecutor,as_completedimportosimportthreadingimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarizefromgatewayimportfetchdefsolve():rows,failures[],[]activepeak0lockthreading.Lock()base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101)withhttpx.Client(trust_envFalse)asclient:deftracked(name):nonlocalactive,peak# 观察整个工作调用的占用不用Future数量冒充活跃线程数。withlock:active1peakmax(peak,active)try:returnfetch(client,base_url,name,normal,2)finally:withlock:active-1withThreadPoolExecutor(max_workers2)aspool:jobs{pool.submit(tracked,name):namefornameinSUPPLIERS}forfutureinas_completed(jobs):try:rows.append(future.result())exceptProviderFailureasexc:failures.append({supplier:jobs[future],code:exc.code})resultsummarize(rows,failures,2)assert[row[supplier]forrowinresult[quotes]][boreal,atlas]tiesummarize([{supplier:boreal,total_cents:100,stock:2},{supplier:atlas,total_cents:100,stock:2}],[],2)assert[row[supplier]forrowintie[quotes]][atlas,boreal]assertactive0andpeak2result[peak_threads]peakreturnresultif__name____main__:print(solve())lessons/07/domain.pyShared business rules copied into each independent lesson.fromdataclassesimportasdict,dataclass SUPPLIERS(atlas,boreal,cedar)classProviderFailure(Exception):def__init__(self,code):self.codecodesuper().__init__(code)classInvalidData(ProviderFailure):def__init__(self):super().__init__(invalid_data)defvalidate_quantity(quantity):在任何外部调用前拒绝非法数量布尔值不能冒充整数。iftype(quantity)isnotintornot1quantity100:raiseValueError(quantity must be an integer from 1 to 100)dataclass(frozenTrue,slotsTrue)classQuote:supplier:strsku:strunit_cents:intshipping_cents:intstock:intclassmethoddefparse(cls,data,supplier,sku):在HTTP边界把不可信JSON变成身份、金额和库存均可信的报价。ifnotisinstance(data,dict):raiseInvalidData()ifdata.get(supplier)!supplierordata.get(sku)!sku:raiseInvalidData()forkeyin(unit_cents,shipping_cents,stock):iftype(data.get(key))isnotintordata[key]0:raiseInvalidData()returncls(supplier,sku,data[unit_cents],data[shipping_cents],data[stock])defrow(self,quantity):return{**asdict(self),total_cents:self.unit_cents*quantityself.shipping_cents}defsummarize(rows,failures,quantity):缺货是正常过滤最终排名由业务键决定与完成顺序无关。quotessorted((rowforrowinrowsifrow[stock]quantity),keylambdarow:(row[total_cents],row[supplier]))return{sku:KB-001,quantity:quantity,quotes:quotes,failures:sorted(failures,keylambdarow:row[supplier]),best:quotes[0]ifquoteselseNone}lessons/07/gateway.pyOne synchronous HTTP attempt; no retry and no batch time budget.fromtimeimportperf_counterimporthttpxfromdomainimportInvalidData,ProviderFailure,Quotedeffetch(client,base_url,supplier,scenario,quantity):只执行一次HTTP尝试客户端归调用者所有此处不创建、不关闭。startedperf_counter()try:responseclient.get(f{base_url}/quotes/{supplier},params{sku:KB-001,scenario:scenario},timeouthttpx.Timeout(0.6))receivedperf_counter()response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure(timeout)fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(fhttp_{exc.response.status_code})fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure(transport)fromexc# HTTP成功不保证数据合法解析错误在供应商边界转成明确类别。try:dataresponse.json()exceptValueErrorasexc:raiseInvalidData()fromexc quoteQuote.parse(data,supplier,KB-001)rowquote.row(quantity)row[http_ms]round((received-started)*1000,2)row[parse_ms]round((perf_counter()-received)*1000,2)returnrowlessons/07/main.pyPure Python CPU work. spawn is explicit on every platform.importargparsefromconcurrent.futuresimportProcessPoolExecutor,as_completedimportjsonimportmultiprocessingimportosfromtimeimportperf_counterdefchecksum(job):纯Python教学校验值不是密码学哈希顶层定义才能被spawn导入。index,sku,roundsjob value0for_inrange(rounds):forcharinsku:value(value*33ord(char))%1000000007return{index:index,sku:sku,checksum:value,pid:os.getpid()}defrun_processes(skus,workers2,rounds80000):只传简单数据跨进程此演示计时包含创建与关闭进程池的成本。contextmultiprocessing.get_context(spawn)withProcessPoolExecutor(max_workersworkers,mp_contextcontext)aspool:futures[pool.submit(checksum,(i,sku,rounds))fori,skuinenumerate(skus)]return[future.result()forfutureinas_completed(futures)]defmain():parserargparse.ArgumentParser()parser.add_argument(--rounds,typeint,default80000)argsparser.parse_args()skus[KB-001,KB-002,KB-003,KB-004]startedperf_counter()serial[checksum((i,sku,args.rounds))fori,skuinenumerate(skus)]serial_ms(perf_counter()-started)*1000startedperf_counter()parallelrun_processes(skus,roundsargs.rounds)parallel_ms(perf_counter()-started)*1000assert[row[checksum]forrowinserial][row[checksum]forrowinsorted(parallel,keylambdarow:row[index])]print(json.dumps({serial_ms:round(serial_ms,2),process_ms:round(parallel_ms,2),completion_order:parallel},ensure_asciiFalse,indent2))# spawn重新导入模块时只能取得函数定义不能递归启动新的进程池。if__name____main__:multiprocessing.freeze_support()main()lessons/07/test_lesson.pyimportunittestimportosfrommainimportchecksum,run_processesclassLessonTests(unittest.TestCase):deftest_spawn_returns_every_job_and_same_checksum_as_serial(self):words[KB-001,KB-002,KB-003]resultrun_processes(words,workers2,rounds1000)self.assertEqual(sorted(row[index]forrowinresult),[0,1,2])self.assertTrue(all(row[pid]!os.getpid()forrowinresult))expected[checksum((i,word,1000))[checksum]fori,wordinenumerate(words)]self.assertEqual([r[checksum]forrinsorted(result,keylambdar:r[index])],expected)if__name____main__:unittest.main()

Python进阶与并发编程07：CPU工作为什么要换进程，以及为什么换完可能更慢

相关推荐

MySQL COUNT函数全解析：三种形态、统计语义与性能优化

SpringBoot整合MyBatis与Thymeleaf实现CRUD实战

2026.9.29：使用 Milvus Operator 在 Kubernetes 中运行 Milvus

更多资讯

AI痕迹太重怎么办？一套可落地的降AI率改稿实操指南

RAG调优实战：从F1 0.42到0.89的30天方法论

Android FTP服务器Demo：手机秒变局域网文件中转站

基于深度学习的人脸识别系统实战：从模型选型到毕设落地

图像生成VR体验：低成本三维重建与全景漫游实战

YOLOv11结合可变形注意力DAT的目标检测改进实践指南

今日更新

NetSpot 实战：用信道分析与频宽优化解决 Wi-Fi 卡顿

C++ split() 字符串分割的五种实现与性能实测

R语言坐标轴范围设置：从xlim/ylim到coord_cartesian的完整指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？