企业官网建设流程全解析

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

报价系统不一定只等待网络。假设我们还要给每个SKU生成一份校验摘要，算法包含大量Python整数运算。这个计算与访问供应商相互独立，可以按商品分开执行。上一课的线程池适合把HTTP等待重叠，但在常见启用GIL的CPython中，它通常不能让多个线程同时执行这段纯Python循环。

本课先构造可重复、无网络的CPU工作，再用进程池执行，观察摘要、进程ID和总耗时。算法只是学习用的滚动校验值，不是密码学哈希，也不能用于密码、数字签名或防篡改。选择它的原因很具体：主要运算发生在Python层，输入很小，输出很小，便于看清并行计算与进程启动两类成本。

E06答案：限制两个线程，不限制结果的确定性

E06要求三个供应商最多占用2个工作线程，并把结果按总价、供应商名称稳定排序。answer_previous.py先创建一个共享同步Client，再创建ThreadPoolExecutor(max_workers=2)。三个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-v

rounds决定每项CPU工作量。2026-10-09本课程制作环境的一次实际运行中，默认80000轮的串行阶段为168.19ms，进程阶段为1885.08ms。这个数字只描述该次Windows/Python3.11.9运行；它已经包含新建进程池、启动spawn工作者、传入任务、等待结果及关闭池，并不是纯计算核耗时。其他机器和后台负载会不同。

输出还可能显示四项任务都由同一个PID完成。max_workers=2表示允许至多两个工作者，不保证短任务一定公平分散；如果一个进程先准备好，另一个还在启动，它可能已经把任务做完。不能写一个测试强制要求“小任务必须出现两个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.py

"""E06: two worker threads and deterministic business ranking."""fromconcurrent.futuresimportThreadPoolExecutor,as_completedimportosimportthreadingimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarizefromgatewayimportfetchdefsolve():rows,failures=[],[]active=peak=0lock=threading.Lock()base_url=os.getenv('QUOTEFLOW_PROVIDER_URL','http://127.0.0.1:9101')withhttpx.Client(trust_env=False)asclient:deftracked(name):nonlocalactive,peak# 观察整个工作调用的占用，不用Future数量冒充活跃线程数。withlock:active+=1peak=max(peak,active)try:returnfetch(client,base_url,name,'normal',2)finally:withlock:active-=1withThreadPoolExecutor(max_workers=2)aspool:jobs={pool.submit(tracked,name):namefornameinSUPPLIERS}forfutureinas_completed(jobs):try:rows.append(future.result())exceptProviderFailureasexc:failures.append({'supplier':jobs[future],'code':exc.code})result=summarize(rows,failures,2)assert[row['supplier']forrowinresult['quotes']]==['boreal','atlas']tie=summarize([{'supplier':'boreal','total_cents':100,'stock':2},{'supplier':'atlas','total_cents':100,'stock':2}],[],2)assert[row['supplier']forrowintie['quotes']]==['atlas','boreal']assertactive==0andpeak<=2result['peak_threads']=peakreturnresultif__name__=='__main__':print(solve())

lessons/07/domain.py

"""Shared business rules copied into each independent lesson."""fromdataclassesimportasdict,dataclass SUPPLIERS=('atlas','boreal','cedar')classProviderFailure(Exception):def__init__(self,code):self.code=codesuper().__init__(code)classInvalidData(ProviderFailure):def__init__(self):super().__init__('invalid_data')defvalidate_quantity(quantity):"""在任何外部调用前拒绝非法数量；布尔值不能冒充整数。"""iftype(quantity)isnotintornot1<=quantity<=100:raiseValueError('quantity must be an integer from 1 to 100')@dataclass(frozen=True,slots=True)classQuote:supplier:strsku:strunit_cents:intshipping_cents:intstock:int@classmethoddefparse(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*quantity+self.shipping_cents}defsummarize(rows,failures,quantity):"""缺货是正常过滤；最终排名由业务键决定，与完成顺序无关。"""quotes=sorted((rowforrowinrowsifrow['stock']>=quantity),key=lambdarow:(row['total_cents'],row['supplier']))return{'sku':'KB-001','quantity':quantity,'quotes':quotes,'failures':sorted(failures,key=lambdarow:row['supplier']),'best':quotes[0]ifquoteselseNone}

lessons/07/gateway.py

"""One synchronous HTTP attempt; no retry and no batch time budget."""fromtimeimportperf_counterimporthttpxfromdomainimportInvalidData,ProviderFailure,Quotedeffetch(client,base_url,supplier,scenario,quantity):"""只执行一次HTTP尝试；客户端归调用者所有，此处不创建、不关闭。"""started=perf_counter()try:response=client.get(f'{base_url}/quotes/{supplier}',params={'sku':'KB-001','scenario':scenario},timeout=httpx.Timeout(0.6))received=perf_counter()response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure('timeout')fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(f'http_{exc.response.status_code}')fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏，不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure('transport')fromexc# HTTP成功不保证数据合法；解析错误在供应商边界转成明确类别。try:data=response.json()exceptValueErrorasexc:raiseInvalidData()fromexc quote=Quote.parse(data,supplier,'KB-001')row=quote.row(quantity)row['http_ms']=round((received-started)*1000,2)row['parse_ms']=round((perf_counter()-received)*1000,2)returnrow

lessons/07/main.py

"""Pure Python CPU work. spawn is explicit on every platform."""importargparsefromconcurrent.futuresimportProcessPoolExecutor,as_completedimportjsonimportmultiprocessingimportosfromtimeimportperf_counterdefchecksum(job):"""纯Python教学校验值，不是密码学哈希；顶层定义才能被spawn导入。"""index,sku,rounds=job value=0for_inrange(rounds):forcharinsku:value=(value*33+ord(char))%1000000007return{'index':index,'sku':sku,'checksum':value,'pid':os.getpid()}defrun_processes(skus,workers=2,rounds=80000):"""只传简单数据跨进程；此演示计时包含创建与关闭进程池的成本。"""context=multiprocessing.get_context('spawn')withProcessPoolExecutor(max_workers=workers,mp_context=context)aspool:futures=[pool.submit(checksum,(i,sku,rounds))fori,skuinenumerate(skus)]return[future.result()forfutureinas_completed(futures)]defmain():parser=argparse.ArgumentParser()parser.add_argument('--rounds',type=int,default=80000)args=parser.parse_args()skus=['KB-001','KB-002','KB-003','KB-004']started=perf_counter()serial=[checksum((i,sku,args.rounds))fori,skuinenumerate(skus)]serial_ms=(perf_counter()-started)*1000started=perf_counter()parallel=run_processes(skus,rounds=args.rounds)parallel_ms=(perf_counter()-started)*1000assert[row['checksum']forrowinserial]==[row['checksum']forrowinsorted(parallel,key=lambdarow:row['index'])]print(json.dumps({'serial_ms':round(serial_ms,2),'process_ms':round(parallel_ms,2),'completion_order':parallel},ensure_ascii=False,indent=2))# spawn重新导入模块时只能取得函数定义，不能递归启动新的进程池。if__name__=='__main__':multiprocessing.freeze_support()main()

lessons/07/test_lesson.py

importunittestimportosfrommainimportchecksum,run_processesclassLessonTests(unittest.TestCase):deftest_spawn_returns_every_job_and_same_checksum_as_serial(self):words=['KB-001','KB-002','KB-003']result=run_processes(words,workers=2,rounds=1000)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,key=lambdar:r['index'])],expected)if__name__=='__main__':unittest.main()