引言
近日，在学习完操作系统的进程调度部分后，我萌生了一个有趣的想法：通过编写代码来模拟进程调度算法，以加深自己对这一知识点的理解。于是，我花了一整天的时间投入到了这个突发奇想的实践中。
背景
进程调度是操作系统中的重要概念，它决定了如何合理地分配处理器时间，以便多个进程能够高效地并发运行。在学习完进程调度算法后，我想通过编写代码来模拟其中一些经典的调度算法，包括先来先服务（FCFS）、最短作业优先（SJF）、轮转调度（Round Robin）、最高响应比（HRRN）和优先级调度。
前言
这里先简单介绍一下我们在进程调度问题里面常见的用语:（以下我用“作业”代表进程和线程）
# 周转时间 = 完成时间- 到达时间
# 带权周转时间 = 周转时间 /运行时间
# 等待时间 = 周转时间- 运行时间
# 平均带权周转时间=带权周转时间 /作业数量
完成时间：操作系统执行这个作业所需的时间
到达时间：作业到达就绪队列的时刻
周转时间：作业被提交给系统开始，到作业完成为止的这段时间间隔
代码实现（每行代码都有注释）
1. 先来先服务 (FCFS)
基本思想
先来先服务的调度算法：最简单的调度算法，系统将按照作业到达的先后次序来进行调度，优先从后备队列中，选择一个或多个位于队列头部的作业，把他们调入内存，分配所需资源、创建进程，然后放入“就绪队列”,直到该进程运行到完成或发生某事件堵塞后，进程调度程序才将处理机分配给其他进程。
实现代码（可直接运行）
# 周转时间 = 完成时间- 到达时间# 带权周转时间 = 周转时间 /运行时间# 等待时间 = 周转时间- 运行时间class Job: # 作业类 def __init__(self, name, arrival_time, burst_time): self.name = name # 作业名 self.arrival_time = arrival_time # 作业到达时间 self.burst_time = burst_time # 作业运行时间def fcfs(Jobs): print("先来先服务算法-----------------") print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") current_time = 0 # 当前时间，记录每次作业运行完后的时间点 total_waiting_time = 0 # 总等待时间 total_turnaround_time_rate = 0 # 总带权周转时间 Jobs.sort(key=lambda x: (x.arrival_time, x.burst_time)) # 按到达时间和执行时间排序 for job in Jobs: if current_time < job.arrival_time: # 作业到达时间比当前时间早(小)，则等待 current_time = job.arrival_time # 更新当前时间为作业到达时间 completion_time = round((current_time + job.burst_time),2) # 作业完成时间=当前时间+作业运行时间 turnaround_time = round((completion_time - job.arrival_time),2) # 作业周转时间=完成时间-到达时间 waiting_time = round((turnaround_time - job.burst_time),2) # 等待时间=周转时间-运行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),) # 带权周转时间=周转时间/运行时间 print( f"{job.name}\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time # 更新当前时间为作业完成时间 total_waiting_time += waiting_time # 累加等待时间 total_turnaround_time_rate += turnaround_time_rate # 累加带权周转时间 print("\n作业总用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)# 示例作业列表Jobs = [ Job("P1", 0, 10), Job("P2", 5, 4), Job("P3", 2, 8), Job("P4", 10, 5), Job("P5", 2, 5)]fcfs(Jobs)2. 最短作业优先 (SJF)
基本思想
当所有作业都到达就绪队列时，这时候操作系统会优先执行所需时间最少的作业（之前我这里理解错了，我以为当有作业0时刻到达，就先执行这个0时刻到达的作业，然后再执行所需时间最短的作业）
实现代码（可直接运行）
#非抢占式的最短作业优先调度算法：#并不是先执行0时刻到达的作业，而是当所有作业队列都到达时，再进行调度class Job: def __init__(self, name, arrival_time, burst_time): self.name = name self.arrival_time = arrival_time self.burst_time = burst_timedef sjf(Jobs): print("最短作业优先算法-----------------") print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") current_time = 0 # 当前时间，记录每次作业运行完后的时间点 total_waiting_time = 0 # 总等待时间 total_turnaround_time_rate = 0 # 总带权周转时间 Jobs.sort(key=lambda x: (x.burst_time, x.arrival_time)) # 按到达时间和运行时间排序 for job in Jobs: if current_time < job.arrival_time: # 当前时间小于作业到达时间，则等待 current_time = job.arrival_time # 当前时间更新为作业到达时间 completion_time = round((current_time + job.burst_time),2) # 作业完成时间=当前时间+作业运行时间 turnaround_time = round((completion_time - job.arrival_time),2) #周转时间=作业完成时间-作业到达时间 waiting_time = round((turnaround_time - job.burst_time),2) #等待时间=周转时间-作业运行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),2) #带权周转时间=周转时间/作业运行时间 print( f"{job.name}\t\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time # 当前时间更新为作业完成时间 total_waiting_time += waiting_time # 总等待时间累加 total_turnaround_time_rate += turnaround_time_rate # 总带权周转时间累加 print("\n作业总用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)# 示例作业列表Jobs = [ Job("P1", 0, 10), Job("P2", 5, 4), Job("P3", 2, 8), Job("P4", 10, 5), Job("P5", 2, 5)]sjf(Jobs)3. 轮转调度 (Round Robin)
基本思想
给每个作业固定的执行时间，根据作业到达的先后顺序让作业在固定的时间片内执行，执行完成后便调度下一个进程执行，时间片轮转调度不考虑进程等待时间和执行时间，属于抢占式调度。优点是兼顾长短作业；缺点是平均等待时间较长，上下文切换较费时。适用于分时系统。
实现代码（可直接运行）
# 时间片轮转算法class Job: def __init__(self, name, arrival_time, burst_time): self.name = name self.arrival_time = arrival_time self.burst_time = burst_timedef round_robin(Jobs, time_slice): print("时间片轮转算法-----------------") print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") current_time = 0 remaining_time = [job.burst_time for job in Jobs] # 用于记录每个作业的剩余执行时间 completed_jobs = [] # 已完成的作业队列 total_waiting_time = 0 # 总等待时间 total_turnaround_time_rate = 0 # 总带权周转时间 while any(remaining > 0 for remaining in remaining_time):# 只要有作业的剩余执行时间大于0，就继续执行 for i, job in enumerate(Jobs):# 遍历作业列表 if remaining_time[i] > 0 and job not in completed_jobs: # 如果作业的剩余执行时间大于0且作业没有完成 if remaining_time[i] <= time_slice: # 如果作业的剩余执行时间小于等于时间片,说明作业可以在本轮执行完 current_time += remaining_time[i] # 当前时间加上作业的剩余执行时间 remaining_time[i] = 0 # 作业的剩余执行时间置为0 completed_jobs.append(job) # 将作业加入已完成的作业队列 else: # 如果作业的剩余执行时间大于时间片,说明作业不能在本轮执行完 current_time += time_slice # 当前时间加上时间片 remaining_time[i] -= time_slice # 作业的剩余执行时间减去时间片 if (len(completed_jobs)==0): continue; for job in completed_jobs: completion_time = round((current_time),2) # 完成时间=当前时间 turnaround_time = round((completion_time - job.arrival_time),2) # 周转时间=完成时间-到达时间 waiting_time = round((turnaround_time - job.burst_time),2) # 等待时间=周转时间-执行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),2) # 带权周转时间=周转时间/执行时间 print( f"{job.name}\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") total_waiting_time += waiting_time # 累加等待时间 total_turnaround_time_rate += turnaround_time_rate # 累加带权周转时间 completed_jobs.remove(job) print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)# 示例作业列表Jobs = [ Job("P1", 0, 4), Job("P2", 1, 3), Job("P3", 2, 4), Job("P4", 3, 2), Job("P5", 4, 4)]time_slice = 4 # 设置时间片大小round_robin(Jobs, time_slice)4.优先级调度算法
基本思想
给每个作业都设置一个优先级，然后在调度的时候，在所有处于就绪状态的任务中选择优先级最高的任务去运行。
实现代码（可直接运行）
#非抢占式优先级调度算法，当所有进程都处于就绪状态时，按照优先级从高到低顺序选择一个进程执行class Job: def __init__(self, name, arrival_time, burst_time, priority): self.name = name self.arrival_time = arrival_time self.burst_time = burst_time self.priority = priority #作业优先级def priority_scheduling(Jobs): print("优先级调度算法-----------------") print("进程\t到达时间\t执行时间\t优先级\t完成时刻\t周转时间\t等待时间\t带权周转时间") current_time = 0 completed_jobs = [] #记录已经完成的作业列表 while Jobs: max_priority = float('-inf') #设置负无穷小为目前的最大优先级 selected_job = None #记录当前选中的作业 for job in Jobs:#选出到达作业中的优先级最高的作业 if job.arrival_time <= current_time and job not in completed_jobs: #如果作业到达时间小于(早于)等于当前时间，并且作业还没有被完成 if job.priority > max_priority:#如果作业优先级大于当前最大优先级，则将当前作业设为选中作业 max_priority = job.priority selected_job = job if selected_job is None: #如果没有选中作业，则说明当前时间没有到达作业，则直接跳过 current_time += 1 continue completion_time = round((current_time + selected_job.burst_time),2) # 完成时间 = 当前时间 + 作业运行时间 turnaround_time = round((completion_time - selected_job.arrival_time),2)#周转时间 = 完成时间 - 到达时间 waiting_time = round((turnaround_time - selected_job.burst_time),2)#等待时间 = 周转时间 - 运行时间 turnaround_time_rate = round((turnaround_time / selected_job.burst_time),2) #带权周转时间 = 周转时间 / 运行时间 print( f"{selected_job.name}\t\t{selected_job.arrival_time}\t\t{selected_job.burst_time}\t\t{selected_job.priority}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time #当前时间更新为完成时间 completed_jobs.append(selected_job) #将选中作业加入已完成作业列表 Jobs.remove(selected_job) #将选中作业从作业列表中移除 total_waiting_time = sum(turnaround_time - job.burst_time for job in completed_jobs)#计算总等待时间 total_turnaround_time_rate = sum(turnaround_time / job.burst_time for job in completed_jobs)#计算总带权周转时间 print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(completed_jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs) print("平均带权周转时间:", avg_turnaround_time_rate)# 示例作业列表Jobs = [ Job("P1", 0, 10, 3), Job("P2", 5, 4, 1), Job("P3", 2, 8, 4), Job("P4", 10, 5, 2), Job("P5", 2, 5, 5)]priority_scheduling(Jobs)5.封装
这里我将四个算法封装在了同一个py文件中，方便用多个测试用例调用这些算法
class Job: def __init__(self, name, arrival_time, burst_time, priority=None): self.name = name self.arrival_time = arrival_time self.burst_time = burst_time self.priority = prioritydef fcfs(Jobs): print("作业-到达时间-服务时间-优先权") for Job in Jobs: print(f"{Job.name}\t{Job.arrival_time}\t{Job.burst_time}\t{Job.priority}") current_time = 0 total_waiting_time = 0 total_turnaround_time_rate = 0 print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") for job in Jobs: if current_time < job.arrival_time: current_time = job.arrival_time completion_time = round((current_time + job.burst_time),2) # 作业完成时间=当前时间+作业运行时间 turnaround_time = round((completion_time - job.arrival_time),2) # 作业周转时间=完成时间-到达时间 waiting_time = round((turnaround_time - job.burst_time),2) # 等待时间=周转时间-运行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),) # 带权周转时间=周转时间/运行时间 print( f"{job.name}\t\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time total_waiting_time += waiting_time total_turnaround_time_rate += turnaround_time_rate print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)def sjf(Jobs): print("作业-到达时间-服务时间-优先权") for Job in Jobs: print(f"{Job.name}\t{Job.arrival_time}\t{Job.burst_time}\t{Job.priority}") Jobs.sort(key=lambda x: (x.burst_time, x.arrival_time)) current_time = 0 total_waiting_time = 0 total_turnaround_time_rate = 0 print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") for job in Jobs: if current_time < job.arrival_time: current_time = job.arrival_time completion_time = round((current_time + job.burst_time),2) # 作业完成时间=当前时间+作业运行时间 turnaround_time = round((completion_time - job.arrival_time),2) #周转时间=作业完成时间-作业到达时间 waiting_time = round((turnaround_time - job.burst_time),2) #等待时间=周转时间-作业运行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),2) #带权周转时间=周转时间/作业运行时间 print( f"{job.name}\t\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time total_waiting_time += waiting_time total_turnaround_time_rate += turnaround_time_rate print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)def rr(Jobs, time_slice): print("时间片轮转算法-----------------") print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") current_time = 0 remaining_time = [job.burst_time for job in Jobs] # 用于记录每个作业的剩余执行时间 completed_jobs = [] # 已完成的作业队列 total_waiting_time = 0 # 总等待时间 total_turnaround_time_rate = 0 # 总带权周转时间 while any(remaining > 0 for remaining in remaining_time):# 只要有作业的剩余执行时间大于0，就继续执行 for i, job in enumerate(Jobs):# 遍历作业列表 if remaining_time[i] > 0 and job not in completed_jobs: # 如果作业的剩余执行时间大于0且作业没有完成 if remaining_time[i] <= time_slice: # 如果作业的剩余执行时间小于等于时间片,说明作业可以在本轮执行完 current_time += remaining_time[i] # 当前时间加上作业的剩余执行时间 remaining_time[i] = 0 # 作业的剩余执行时间置为0 completed_jobs.append(job) # 将作业加入已完成的作业队列 else: # 如果作业的剩余执行时间大于时间片,说明作业不能在本轮执行完 current_time += time_slice # 当前时间加上时间片 remaining_time[i] -= time_slice # 作业的剩余执行时间减去时间片 if (len(completed_jobs)==0): continue; for job in completed_jobs: completion_time = round((current_time),2) # 完成时间=当前时间 turnaround_time = round((completion_time - job.arrival_time),2) # 周转时间=完成时间-到达时间 waiting_time = round((turnaround_time - job.burst_time),2) # 等待时间=周转时间-执行时间 turnaround_time_rate = round((turnaround_time / job.burst_time),2) # 带权周转时间=周转时间/执行时间 print( f"{job.name}\t{job.arrival_time}\t\t{job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") total_waiting_time += waiting_time # 累加等待时间 total_turnaround_time_rate += turnaround_time_rate # 累加带权周转时间 completed_jobs.remove(job) print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(Jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs) print("平均带权周转时间:", avg_turnaround_time_rate)def priority_scheduling(Jobs): print("作业-到达时间-服务时间-优先权") for Job in Jobs: print(f"{Job.name}\t{Job.arrival_time}\t{Job.burst_time}\t{Job.priority}") current_time = 0 completed_jobs = [] print("进程\t到达时间\t执行时间\t优先级\t完成时刻\t周转时间\t等待时间\t带权周转时间") while Jobs: max_priority = float('-inf') selected_job = None for job in Jobs: if job.arrival_time <= current_time and job not in completed_jobs: if job.priority > max_priority: max_priority = job.priority selected_job = job if selected_job is None: current_time += 1 continue completion_time = round((current_time + selected_job.burst_time),2) # 完成时间 = 当前时间 + 作业运行时间 turnaround_time = round((completion_time - selected_job.arrival_time),2)#周转时间 = 完成时间 - 到达时间 waiting_time = round((turnaround_time - selected_job.burst_time),2)#等待时间 = 周转时间 - 运行时间 turnaround_time_rate = round((turnaround_time / selected_job.burst_time),2) #带权周转时间 = 周转时间 / 运行时间 print( f"{selected_job.name}\t\t{selected_job.arrival_time}\t\t{selected_job.burst_time}\t\t{selected_job.priority}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time completed_jobs.append(selected_job) Jobs.remove(selected_job) total_waiting_time = sum(turnaround_time - job.burst_time for job in completed_jobs) total_turnaround_time_rate = sum(turnaround_time / job.burst_time for job in completed_jobs) print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(completed_jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs) print("平均带权周转时间:", avg_turnaround_time_rate)def hrrn(Jobs): print("作业-到达时间-服务时间-优先权") for Job in Jobs: print(f"{Job.name}\t{Job.arrival_time}\t{Job.burst_time}\t{Job.priority}") current_time = 0 completed_jobs = [] print("进程\t到达时间\t执行时间\t完成时刻\t周转时间\t等待时间\t带权周转时间") while Jobs: max_response_ratio = -1 selected_job = None for job in Jobs: if job.arrival_time <= current_time and job not in completed_jobs: response_ratio = (current_time - job.arrival_time + job.burst_time) / job.burst_time if response_ratio > max_response_ratio: max_response_ratio = response_ratio selected_job = job if selected_job is None: current_time += 1 continue completion_time = round((current_time + selected_job.burst_time),2) #计算完成时间 turnaround_time = round((completion_time - selected_job.arrival_time),2) #计算周转时间 waiting_time = round((turnaround_time - selected_job.burst_time),2) #计算等待时间 turnaround_time_rate =round((turnaround_time / selected_job.burst_time),2) #计算带权周转时间 print( f"{selected_job.name}\t\t{selected_job.arrival_time}\t\t{selected_job.burst_time}\t\t{completion_time}\t\t{turnaround_time}\t\t{waiting_time}\t\t{turnaround_time_rate:.2f}") current_time = completion_time completed_jobs.append(selected_job) Jobs.remove(selected_job) total_waiting_time = sum(turnaround_time - job.burst_time for job in completed_jobs) total_turnaround_time_rate = sum(turnaround_time / job.burst_time for job in completed_jobs) print("\n用时:", current_time) print("平均等待时间:", total_waiting_time / len(completed_jobs)) avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs) print("平均带权周转时间:", avg_turnaround_time_rate)# 作业列表Jobs = [ Job("P1", 0, 4,0), Job("P2", 1, 3,0), Job("P3", 2, 5,0), Job("P4", 3, 2,0), Job("P5", 4, 4,0)]Jobs2 = [ Job("P1", 0, 10, 3), Job("P2", 1, 13, 1), Job("P3", 2, 8, 4), Job("P4", 3, 9, 2), Job("P5", 2, 7, 5)] print("先来先服务算法-------------------------------------------------------------------------------------------------------------------------------")fcfs(Jobs.copy())#浅拷贝，防止原列表被修改print("\n最短作业优先算法-------------------------------------------------------------------------------------------------------------------------------")sjf(Jobs.copy())print("\n时间片轮转算法-------------------------------------------------------------------------------------------------------------------------------")time_slice = 3rr(Jobs.copy(), time_slice)print("\n优先级调度算法-------------------------------------------------------------------------------------------------------------------------------")priority_scheduling(Jobs.copy())print("\n高响应比优先算法-----------------------------------------------------------------------------------------------------------------------------")hrrn(Jobs.copy())运行结果
总结
通过一天的努力，我成功地编写了用代码模拟进程调度算法的示例，包括先来先服务、最短作业优先、轮转调度和优先级调度算法。虽然这个实践实现的东西很简单，还有很多实际问题、特殊情况没有考虑到，但是这个过程不仅加深了我对操作系统进程调度算法的理解，还让我更深入地体验了算法在实际应用中的工作原理。我希望这种实践能够帮助我、同时帮助你们更好地掌握这一重要概念。
写这篇文章是为了记录我今天的成果，当然以后我肯定用得上这些代码，毕竟我还没学操作系统这门课，下学期才学，我相信肯定有要我们算什么“周转时间”、“平均等待时间”的这些题目，大伙如果碰到了这种题目，可以用代码来检测一下自己的结果是不是正确（当然我写的示例过于简单，实际的调度比这个复杂了多，有些特殊情况可能没有考虑到）哈哈哈，如果你觉得我说的没错就给我点个赞呗！
来源地址：https://blog.csdn.net/weixin_46713492/article/details/132263000
