简介这是一份围绕Spooling假脱机输入输出模拟技术的操作系统课程设计资料包面向正在学习设备管理、进程同步与虚拟设备机制的高校学生及课程实验者帮助其把抽象概念落到可运行的程序实现中。包内共11个文件以Java源程序与编译后的class文件为主体辅以一份设计报告doc和4张运行截图jpg压缩包约279KB规模轻量但结构完整。目前已有442人学习适合用作课程设计参考或实验复盘。读者可结合源代码观察请求输出进程与Spooling输出进程如何协作理解输入井、输出井和后台处理程序在缓冲调度中的职责并借助报告梳理设计思路、实现方法与性能分析截图则呈现初始化、数据输入、文件为空及运行界面等关键状态便于对照程序行为、定位问题并加深对共享慢速I/O设备优化思路的理解。1. 从打印室排队到磁盘井spooling假脱机输入输出模拟技术到底在模拟什么spooling假脱机输入输出模拟技术本质上是在用磁盘上的缓冲区把一台独占设备伪装成多用户共享设备。某公司打印室门口贴过一张纸每人先把文件提交到一个目录再由一台机器按队列出纸。很多人以为这就是排队真正写模拟器时才发现难点不在队列而在“输入井、输出井、输入进程、输出进程”四者之间的状态同步和容量控制。这个方向适合操作系统课程设计、系统编程练手也适合需要理解缓冲、调度和并发控制的工程师。下面从四个对象的文件系统落地讲起再给一个能跑的Python骨架和参数边界。2. 输入井和输出井怎么落地用文件系统模拟磁盘缓冲区的四个关键对象2.1 输入井、输出井、输入进程、输出进程分别对应什么SPOOLing系统里输入井和输出井是磁盘上的缓冲区域输入进程负责把慢速输入设备的数据提前读入输入井输出进程负责把输出井里的数据按设备节奏送出。用户进程不再直接碰独占设备只和井打交道于是打印机这类设备就从“一次只能一个进程用”变成“多个进程可以先后提交、由系统统一调度”。模拟技术要做的是把这套抽象映射到普通文件系统目录当井文件当块JSON当作业控制块线程当外围进程。我在模拟项目X里一般会先画一张映射表把每个概念落到可检查的对象上否则后面排查时很容易把“井满”和“队列满”混在一起。真实系统对象模拟实现关键参数排查时看什么输入井input_well/目录井容量、块大小目录总字节数、作业状态输出井output_well/目录井容量、并发度是否有PRINTING卡住输入进程一个扫描输入井的线程扫描间隔、批量大小是否长期空转输出进程一个或多个写sink的线程设备锁、块间隔sink是否交错井管理程序Spooler类调度策略、超时状态机是否闭环作业控制块meta.jsonjob_id、blocks、stateJSON是否完整这张表的价值在于每个参数都能在目录里找到对应证据。比如“井满”不是看内存计数器而是看input_well和output_well下*.bin的总大小比如“输出进程卡住”不是猜线程而是看meta.json里状态是否停在PRINTING。2.2 用目录树和元数据模拟“井”结构设计目录结构不要平铺一个作业一个目录作业目录里再放块文件和元数据。这样移动作业时只需要一次目录重命名比逐个文件搬移更接近“作业整体入井、整体出井”的语义。下面是我常用的最小结构spool_root是模拟根目录sink/printer.out代表最终输出设备。spool_root/ ├── input_well/ # 输入井等待输入进程处理的作业 ├── output_well/ # 输出井等待输出进程送到设备的作业 ├── done/ # 已完成作业保留现场便于校验 ├── tmp/ # 提交中的临时目录保证原子性 ├── sink/ │ └── printer.out # 模拟独占输出设备 └── source_jobs/ # 用户提交的源文件仅用于演示每个作业目录里长这样input_well|output_well|done/job_id/ ├── meta.json ├── chunk_0000.bin ├── chunk_0001.bin └── ...meta.json至少要有job_id、name、size、block_size、blocks、priority、state、submitted_at、owner这些字段。状态机建议固定成IN_WELL - OUTPUT_WELL - PRINTING - DONE失败时进FAILED。状态不要只放在内存里因为模拟器一重启内存状态就没了而井里的文件还在把状态写进元数据才能做恢复和排查。注意meta.json不要最后才写。一个常见做法是先把块文件写进tmp/job_id/最后写meta.json再用一次目录重命名提交到input_well/。这样输出进程永远不会看到“只有一半块”的作业。2.3 从独占到共享一次作业在井里的完整路径一次提交的路径可以拆成五步。第一步用户进程把源文件读成字节流按block_size切块写入tmp/job_id/chunk_XXXX.bin同时生成元数据。第二步tmp/job_id原子重命名到input_well/job_id此时作业才算真正入井。第三步输入进程按调度策略从输入井取作业把状态改成OUTPUT_WELL再把整个目录移动到输出井这一步模拟的是“输入设备的数据已经进入缓冲区等待输出设备消费”。第四步输出进程从输出井取作业把状态改成PRINTING按块顺序写入sink/printer.out。第五步写完后状态改成DONE目录移到done/同时释放井容量。这里有个容易被忽略的细节输入井和输出井都可以叫“井”但它们的容量统计口径要一致。我一般只统计*.bin的字节数不把meta.json算进去因为元数据开销相对固定混进去会让“井大小”这个参数变得不可解释。调度时也不要只按文件名排序文件名是随机ID排序没有业务含义应该按priority和submitted_at组合排序优先级数字小的先出同优先级按提交时间先入先出。3. 用Python跑通最小SPOOLing模拟器从submit到假脱机输出的可运行骨架3.1 环境准备与目录初始化环境只需要Python标准库不依赖第三方包。下面命令创建演示目录source_jobs放几个待提交文件spool_root是模拟器的根。运行前把旧目录删掉避免上一次的done/干扰观察。mkdir -p spool_demo/source_jobs cd spool_demo python3 --version代码放在spool_demo/spooler_demo.py。核心类叫Spooler它负责容量控制、原子提交、输入进程和输出进程。下面这段代码可以直接运行重点看submit里的cv.wait和os.replace以及output_worker里的sink_lock。import os import json import time import uuid import threading from pathlib import Path from dataclasses import dataclass, asdict dataclass class JobMeta: job_id: str name: str size: int block_size: int blocks: int priority: int state: str submitted_at: float owner: str class Spooler: def __init__(self, root, block_size4096, max_well_bytes1024 * 1024): self.root Path(root) self.block_size block_size self.max_well_bytes max_well_bytes self.input_well self.root / input_well self.output_well self.root / output_well self.done self.root / done self.tmp self.root / tmp self.sink self.root / sink / printer.out for p in [self.input_well, self.output_well, self.done, self.tmp, self.sink.parent]: p.mkdir(parentsTrue, exist_okTrue) self.lock threading.Lock() self.cv threading.Condition(self.lock) self.sink_lock threading.Lock() self.stop_flag False self.used_bytes self._well_used() def _well_used(self): total 0 for well in (self.input_well, self.output_well): for f in well.rglob(*.bin): total f.stat().st_size return total def submit(self, src, priority5, owneruser): src Path(src) data src.read_bytes() size len(data) with self.cv: while self.used_bytes size self.max_well_bytes and not self.stop_flag: self.cv.wait(timeout0.5) if self.stop_flag: raise RuntimeError(spooler stopped) job_id uuid.uuid4().hex[:12] tmp_job self.tmp / job_id tmp_job.mkdir(parentsTrue) blocks 0 for off in range(0, size, self.block_size): chunk data[off:off self.block_size] (tmp_job / fchunk_{blocks:04d}.bin).write_bytes(chunk) blocks 1 meta JobMeta(job_id, src.name, size, self.block_size, blocks, priority, IN_WELL, time.time(), owner) (tmp_job / meta.json).write_text( json.dumps(asdict(meta), ensure_asciiFalse), encodingutf-8) os.replace(tmp_job, self.input_well / job_id) # 原子提交 self.used_bytes size self.cv.notify_all() return job_id def _pick_job(self, well, expected_state): jobs [] for d in well.iterdir(): if not d.is_dir(): continue mp d / meta.json if not mp.exists(): continue meta JobMeta(**json.loads(mp.read_text(encodingutf-8))) if meta.state expected_state: jobs.append((meta.priority, meta.submitted_at, d, meta)) if not jobs: return None jobs.sort(keylambda x: (x[0], x[1])) return jobs[0][3], jobs[0][2] def input_worker(self): while not self.stop_flag: with self.cv: picked self._pick_job(self.input_well, IN_WELL) if not picked: self.cv.wait(timeout0.2) continue meta, d picked meta.state OUTPUT_WELL (d / meta.json).write_text( json.dumps(asdict(meta), ensure_asciiFalse), encodingutf-8) os.replace(d, self.output_well / meta.job_id) self.cv.notify_all() def output_worker(self): while not self.stop_flag: with self.cv: picked self._pick_job(self.output_well, OUTPUT_WELL) if not picked: self.cv.wait(timeout0.2) continue meta, d picked meta.state PRINTING (d / meta.json).write_text( json.dumps(asdict(meta), ensure_asciiFalse), encodingutf-8) with self.sink_lock: with self.sink.open(ab) as out: for i in range(meta.blocks): chunk (d / fchunk_{i:04d}.bin).read_bytes() out.write(chunk) out.flush() time.sleep(0.01) # 模拟慢速设备 with self.cv: meta.state DONE (d / meta.json).write_text( json.dumps(asdict(meta), ensure_asciiFalse), encodingutf-8) os.replace(d, self.done / meta.job_id) self.used_bytes - meta.size self.cv.notify_all()这段代码的逻辑说明分三点。第一submit在条件变量里等待井容量used_bytes只统计*.bin避免元数据把容量算乱等待带timeout0.5是为了让停止标志能被及时看到。第二os.replace(tmp_job, input_well/job_id)是原子提交的关键输出进程要么看不到目录要么看到完整目录不会读到半个作业。第三output_worker在锁内把状态改成PRINTING再释放锁去写sink这样另一个输出线程不会重复取同一个作业sink_lock包住整个作业的写入避免两个作业的块交错。参数方面block_size4096对应常见的4KB块块越小元数据文件越多块越大单次写设备的时间越长max_well_bytes1024*1024是1MB井容量演示够用真实场景按磁盘配额和设备速率调。priority数字越小越先出submitted_at用来在同优先级下做先入先出。3.2 启动生产者和消费者运行与观察入口脚本负责创建源文件、启动输入进程和两个输出进程、提交作业然后等几秒停止。两个输出进程会争sink_lock所以输出内容不会交错但完成顺序受调度和锁竞争影响。if __name__ __main__: sp Spooler(./spool_root, block_size4096, max_well_bytes1024 * 1024) threading.Thread(targetsp.input_worker, daemonTrue).start() for _ in range(2): threading.Thread(targetsp.output_worker, daemonTrue).start() src_dir Path(./source_jobs) src_dir.mkdir(exist_okTrue) for i in range(5): (src_dir / fjob_{i}.txt).write_text( fuser job {i}\n (A * 5000), encodingutf-8) for p in sorted(src_dir.glob(*.txt)): jid sp.submit(p, priority5) print(submitted, jid, p.name) time.sleep(3) sp.stop_flag True with sp.cv: sp.cv.notify_all()运行python3 spooler_demo.py后先看spool_root的目录变化input_well会短暂出现作业随后移动到output_well最后进done。再看spool_root/sink/printer.out的大小应该等于5个源文件的字节数之和。如果大小对不上优先查output_well里有没有卡在PRINTING的目录以及done里meta.json的blocks和实际块文件数是否一致。python3 spooler_demo.py find spool_root -maxdepth 2 -type d | sort wc -c spool_root/sink/printer.out cat spool_root/sink/printer.out | head -c 100这里有个观察技巧不要只看最终sink文件done/里的元数据才是“黑匣子”。每个作业的state应该是DONEblocks应该等于ceil(size/block_size)。如果某个作业的state是PRINTING但进程已经停了说明输出线程在写sink时异常退出或者stop_flag在写块中途被置位。模拟器不是生产系统但状态机闭环这件事必须一开始就做对。4. 参数怎么调井大小、块大小、并发度与调度策略的边界4.1 井大小和块大小不是越大越好max_well_bytes决定同时能容纳多少字节的作业。设得太小提交线程会频繁等待吞吐上不去设得太大磁盘占用和恢复时间都会增加而且“井”失去了缓冲的意义。我一般先按“设备一分钟能输出多少字节”估算再乘以2到3倍作为井容量。比如模拟打印机每秒写100KB井容量设20MB到30MB就够不需要把整个磁盘都当井。块大小的影响更隐蔽。块小元数据文件多目录遍历和打开文件次数增加块大单次写设备的时间变长输出进程持锁时间变长其他作业等待更久。4KB到64KB是常见区间演示用4KB压测时可以试16KB和64KB。下面这组参数是我在模拟项目X里会先跑的对照不是绝对标准但能快速看出拐点。参数建议起点调大后的变化调小后的变化block_size4KB~16KB元数据少单次写久锁占用长元数据多目录遍历慢max_well_bytes设备速率×2~3抗突发强磁盘占用高提交等待多吞吐低input_worker数量1多线程争目录锁收益低不适用output_worker数量1~2受sink锁限制多了会等设备利用率低priority范围0~9数字小先出易饥饿范围窄区分度低提示井容量统计一定要和清理逻辑成对出现。submit加了多少字节作业完成时就要减多少字节异常路径也要减否则跑几十个作业后“井空但提交被拒”就是必然。4.2 并发度输入进程和输出进程各开几个输入进程的工作是扫描目录、改状态、移动目录瓶颈在目录锁和文件系统元数据操作不在CPU。开多个输入线程它们会争同一把self.cv锁收益很小反而增加状态竞争。我一般只开一个输入进程把扫描间隔控制在0.1秒到0.5秒之间间隔太短会空转烧CPU太长会让作业在输入井里多躺一会儿。输出进程的数量要看设备模型。真实打印机是串行设备同一时刻只能打一份模拟器里用sink_lock包住整个作业开两个输出线程也只是让第二个线程在锁外等并不会提高设备吞吐。真正能提高吞吐的是“把多个小作业合并成一次设备启动”这属于调度技巧不是加线程能解决的。压测时如果发现output_worker增加后总时间没降先看sink_lock的等待时间而不是继续加线程。4.3 调度策略FCFS、优先级、短作业优先在井里的差别代码里的_pick_job用(priority, submitted_at)排序等价于“优先级优先同级先入先出”。这对演示够用但真实井里还要考虑作业大小。短作业优先能降低平均等待时间但大作业可能饥饿优先级调度能保证重要作业先出但低优先级作业可能一直排不上。常见做法是加老化每等一轮优先级数字减一减到0为止。策略排序键优点风险FCFSsubmitted_at实现简单公平大作业堵住后面小作业优先级(priority, submitted_at)重要作业先出低优先级饥饿短作业优先(blocks, submitted_at)平均等待短大作业可能永远排不上优先级老化(priority - age, submitted_at)兼顾重要性和公平老化参数要调我在模拟项目X里踩过一个坑把priority设成0到9但提交时全用默认5结果调度看起来像FCFS误以为优先级没生效。后来在提交入口按作业类型分配优先级才看出差别。参数不是设了就有用得让输入数据真的落到不同取值上。5. 避坑与排查spooling假脱机模拟里最常见的五类翻车5.1 井满后提交线程一直等待条件变量没有超时和回滚现象是submit调用后进程不动日志停在“waiting for well space”。原因通常有两个cv.wait()没写超时或者used_bytes只增不减。前者在停止标志置位后仍然醒不过来后者会让容量永远回不到零。解决方式是把等待改成cv.wait(timeout0.5)每次醒来检查stop_flag和容量提交失败时把已经写入tmp的目录删掉不能留下半成品占空间。更稳妥的做法是把“申请容量”和“写块”放在同一个try/finally里异常时统一回滚。5.2 作业目录只写了一半就被输出进程看到缺少原子提交现象是输出进程报FileNotFoundError或JSONDecodeError或者sink文件比预期短。原因是提交线程直接往input_well/job_id里写块输出进程扫描时看到了不完整目录。解决方式是先写tmp/job_id最后写meta.json再用os.replace整体重命名到input_well。如果文件系统不支持目录原子重命名就用一个ready标记文件输出进程只处理带ready的目录。原子提交这件事没有后悔药必须在第一版就做。5.3 两个输出进程重复打印同一个作业状态更新不在锁内现象是printer.out里同一份内容出现两次done/里却只有一个作业目录。原因是输出进程在锁外选作业两个线程同时看到OUTPUT_WELL都认为自己拿到了。解决方式是把“选作业改状态为PRINTING”放在同一个锁内改完再释放锁去写sink或者用os.replace把作业目录先移动到printing/私有目录谁移动成功谁负责。状态机的每一次迁移都要有互斥保护不能靠“应该不会同时发生”来赌。5.4 删除作业后used_bytes不归零统计口径和清理顺序错现象是井目录已经空了但新作业提交仍然被拒打印used_bytes发现是个正数。原因通常是减容量写在os.replace之前移动失败后没有回滚或者异常路径直接return跳过了释放。解决方式是封装一个release_job(meta)在作业进入DONE或FAILED时统一调用内部先减容量再通知条件变量。统计口径也要固定只减*.bin的字节数不要把meta.json算进去否则加减不对称。5.5 输出文件交错sink文件没有按作业加锁现象是printer.out里作业A的前半段和作业B的后半段穿插。原因是多个输出线程各自打开sink追加写块级写入不是原子的。解决方式是用sink_lock包住整个作业的写入循环而不是只包住单次write或者每个作业先写到done/job_id.out最后再按完成顺序合并。真实设备本来就是串行的模拟器里让输出进程排队等设备锁反而更接近真实语义。6. 从能跑到可信spooling模拟器的压测、日志校验与一个调度技巧6.1 压测脚本用随机大小和优先级跑100个作业演示跑5个作业看不出容量和调度的边界。我一般写一个压测入口随机生成1KB到200KB的文件优先级在0到9之间随机提交100个作业后等所有作业进done再校验sink总字节数是否等于源文件总字节数。下面脚本可以放在同一个目录复用Spooler类。import random import time from pathlib import Path def stress(sp, n100): src_dir Path(./stress_jobs) src_dir.mkdir(exist_okTrue) total 0 for i in range(n): size random.randint(1024, 200 * 1024) data bytes([random.randint(65, 90)]) * size p src_dir / fstress_{i:03d}.bin p.write_bytes(data) total size sp.submit(p, priorityrandom.randint(0, 9)) deadline time.time() 30 while time.time() deadline: done len(list(sp.done.iterdir())) if done n: break time.sleep(0.2) sink_size sp.sink.stat().st_size print(expected, total, sink, sink_size, done, len(list(sp.done.iterdir()))) assert sink_size total, sink size mismatch这段脚本的关键不是“跑完”而是最后那个断言。sink总字节数必须等于所有源文件字节数之和少了说明有作业丢块多了说明重复输出。压测时把max_well_bytes调到刚好容纳十几个作业观察submit的等待次数再把block_size从4KB改到64KB比较完成时间。日志里至少记录job_id、state、时间戳和字节数出问题时按job_id重放状态机比翻控制台输出快得多。6.2 一个调度技巧给输出井加“设备空闲窗口”最后说一个我后来才想明白的技巧。输出进程不必来一个作业就立刻启动设备可以在输出井里等一个很短的“空闲窗口”比如20毫秒把窗口内到达的小作业按优先级合并成一批再一次性写sink。这样设备启动次数少了小作业的等待时间也不会明显变长。实现上就是在output_worker取作业前加一个time.sleep(0.02)的窗口等待或者用一个批量队列攒够N个块再写。参数别设太大窗口超过100毫秒交互型作业的延迟就会变得明显。我现在的习惯是先把状态机和原子提交做对再谈参数和调度。spooling假脱机输入输出模拟技术看起来像操作系统课里的老概念但真落到文件、锁、条件变量和异常回滚上坑一个不少。每次调参前先看done/里的元数据比盯着sink文件猜原因可靠得多。希望帮到你。本文还有配套的精品资源点击获取