?引言
近日，在學(xué)習(xí)完操作系統(tǒng)的進(jìn)程調(diào)度部分后，我萌生了一個(gè)有趣的想法：通過(guò)編寫(xiě)代碼來(lái)模擬進(jìn)程調(diào)度算法，以加深自己對(duì)這一知識(shí)點(diǎn)的理解。于是，我花了一整天的時(shí)間投入到了這個(gè)突發(fā)奇想的實(shí)踐中。
?背景
進(jìn)程調(diào)度是操作系統(tǒng)中的重要概念，它決定了如何合理地分配處理器時(shí)間，以便多個(gè)進(jìn)程能夠高效地并發(fā)運(yùn)行。在學(xué)習(xí)完進(jìn)程調(diào)度算法后，我想通過(guò)編寫(xiě)代碼來(lái)模擬其中一些經(jīng)典的調(diào)度算法，包括先來(lái)先服務(wù)（FCFS）、最短作業(yè)優(yōu)先（SJF）、輪轉(zhuǎn)調(diào)度（Round Robin）、最高響應(yīng)比（HRRN）和優(yōu)先級(jí)調(diào)度。
前言
這里先簡(jiǎn)單介紹一下我們?cè)谶M(jìn)程調(diào)度問(wèn)題里面常見(jiàn)的用語(yǔ):（以下我用“作業(yè)”代表進(jìn)程和線(xiàn)程）
# 周轉(zhuǎn)時(shí)間 ? ? = 完成時(shí)間- 到達(dá)時(shí)間
# 帶權(quán)周轉(zhuǎn)時(shí)間 ?= 周轉(zhuǎn)時(shí)間 /運(yùn)行時(shí)間
# 等待時(shí)間 ? ? = 周轉(zhuǎn)時(shí)間- 運(yùn)行時(shí)間
# 平均帶權(quán)周轉(zhuǎn)時(shí)間=帶權(quán)周轉(zhuǎn)時(shí)間 /作業(yè)數(shù)量
完成時(shí)間：操作系統(tǒng)執(zhí)行這個(gè)作業(yè)所需的時(shí)間
到達(dá)時(shí)間：作業(yè)到達(dá)就緒隊(duì)列的時(shí)刻
周轉(zhuǎn)時(shí)間：作業(yè)被提交給系統(tǒng)開(kāi)始，到作業(yè)完成為止的這段時(shí)間間隔
代碼實(shí)現(xiàn)（每行代碼都有注釋?zhuān)?/h4>
1. 先來(lái)先服務(wù) (FCFS)
基本思想
先來(lái)先服務(wù)的調(diào)度算法：最簡(jiǎn)單的調(diào)度算法，系統(tǒng)將按照作業(yè)到達(dá)的先后次序來(lái)進(jìn)行調(diào)度，優(yōu)先從后備隊(duì)列中，選擇一個(gè)或多個(gè)位于隊(duì)列頭部的作業(yè)，把他們調(diào)入內(nèi)存，分配所需資源、創(chuàng)建進(jìn)程，然后放入“就緒隊(duì)列”,直到該進(jìn)程運(yùn)行到完成或發(fā)生某事件堵塞后，進(jìn)程調(diào)度程序才將處理機(jī)分配給其他進(jìn)程。
實(shí)現(xiàn)代碼（可直接運(yùn)行）
# 周轉(zhuǎn)時(shí)間 = 完成時(shí)間- 到達(dá)時(shí)間
# 帶權(quán)周轉(zhuǎn)時(shí)間 = 周轉(zhuǎn)時(shí)間 /運(yùn)行時(shí)間
# 等待時(shí)間 = 周轉(zhuǎn)時(shí)間- 運(yùn)行時(shí)間
class Job: # 作業(yè)類(lèi)
def __init__(self, name, arrival_time, burst_time):
self.name = name # 作業(yè)名
self.arrival_time = arrival_time # 作業(yè)到達(dá)時(shí)間
self.burst_time = burst_time # 作業(yè)運(yùn)行時(shí)間
def fcfs(Jobs):
print("先來(lái)先服務(wù)算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0 # 當(dāng)前時(shí)間，記錄每次作業(yè)運(yùn)行完后的時(shí)間點(diǎn)
total_waiting_time = 0 # 總等待時(shí)間
total_turnaround_time_rate = 0 # 總帶權(quán)周轉(zhuǎn)時(shí)間
Jobs.sort(key=lambda x: (x.arrival_time, x.burst_time)) # 按到達(dá)時(shí)間和執(zhí)行時(shí)間排序
for job in Jobs:
if current_time < job.arrival_time: # 作業(yè)到達(dá)時(shí)間比當(dāng)前時(shí)間早(小)，則等待
current_time = job.arrival_time # 更新當(dāng)前時(shí)間為作業(yè)到達(dá)時(shí)間
completion_time = round((current_time + job.burst_time),2) # 作業(yè)完成時(shí)間=當(dāng)前時(shí)間+作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) # 作業(yè)周轉(zhuǎn)時(shí)間=完成時(shí)間-到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) # 等待時(shí)間=周轉(zhuǎn)時(shí)間-運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),) # 帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/運(yùn)行時(shí)間
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 # 更新當(dāng)前時(shí)間為作業(yè)完成時(shí)間
total_waiting_time += waiting_time # 累加等待時(shí)間
total_turnaround_time_rate += turnaround_time_rate # 累加帶權(quán)周轉(zhuǎn)時(shí)間
print("\n作業(yè)總用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
# 示例作業(yè)列表
Jobs = [
Job("P1", 0, 10),
Job("P2", 5, 4),
Job("P3", 2, 8),
Job("P4", 10, 5),
Job("P5", 2, 5)
]
fcfs(Jobs)
2. 最短作業(yè)優(yōu)先 (SJF)
基本思想
當(dāng)所有作業(yè)都到達(dá)就緒隊(duì)列時(shí)，這時(shí)候操作系統(tǒng)會(huì)優(yōu)先執(zhí)行所需時(shí)間最少的作業(yè)（之前我這里理解錯(cuò)了，我以為當(dāng)有作業(yè)0時(shí)刻到達(dá)，就先執(zhí)行這個(gè)0時(shí)刻到達(dá)的作業(yè)，然后再執(zhí)行所需時(shí)間最短的作業(yè)）
實(shí)現(xiàn)代碼（可直接運(yùn)行）
#非搶占式的最短作業(yè)優(yōu)先調(diào)度算法：
#先執(zhí)行0時(shí)刻到達(dá)的執(zhí)行時(shí)間最短的作業(yè)
class Job:
def __init__(self, name, arrival_time, burst_time):
self.name = name
self.arrival_time = arrival_time
self.burst_time = burst_time
def sjf(Jobs):
print("最短作業(yè)優(yōu)先算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0 # 當(dāng)前時(shí)間，記錄每次作業(yè)運(yùn)行完后的時(shí)間點(diǎn)
total_waiting_time = 0 # 總等待時(shí)間
total_turnaround_time_rate = 0 # 總帶權(quán)周轉(zhuǎn)時(shí)間
Jobs.sort(key=lambda x: (x.burst_time, x.arrival_time)) # 按到達(dá)時(shí)間和運(yùn)行時(shí)間排序
#for job in Jobs:
# print(job.name)
#這里的for循環(huán)，判斷是否有0時(shí)刻到達(dá)的作業(yè)，將0時(shí)刻到達(dá)的執(zhí)行時(shí)間最短的作業(yè)放到隊(duì)列首部先執(zhí)行
for job in Jobs:
if job.arrival_time==0:
Jobs.remove(job)
Jobs.insert(0,job) #插入到隊(duì)首
break #上面已經(jīng)按照運(yùn)行時(shí)間拍好序了，所以這里找到的作業(yè)是所有0時(shí)刻到達(dá)的作業(yè)中執(zhí)行時(shí)間最短的，直接退出循環(huán)
for job in Jobs:
if current_time < job.arrival_time: # 當(dāng)前時(shí)間小于作業(yè)到達(dá)時(shí)間，則等待
current_time = job.arrival_time # 當(dāng)前時(shí)間更新為作業(yè)到達(dá)時(shí)間
completion_time = round((current_time + job.burst_time),2) # 作業(yè)完成時(shí)間=當(dāng)前時(shí)間+作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) #周轉(zhuǎn)時(shí)間=作業(yè)完成時(shí)間-作業(yè)到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) #等待時(shí)間=周轉(zhuǎn)時(shí)間-作業(yè)運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),2) #帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/作業(yè)運(yùn)行時(shí)間
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 # 當(dāng)前時(shí)間更新為作業(yè)完成時(shí)間
total_waiting_time += waiting_time # 總等待時(shí)間累加
total_turnaround_time_rate += turnaround_time_rate # 總帶權(quán)周轉(zhuǎn)時(shí)間累加
print("\n作業(yè)總用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
# 示例作業(yè)列表
Jobs = [
Job("P1", 0, 10),
Job("P2", 5, 4),
Job("P3", 2, 8),
Job("P4", 10, 5),
Job("P5", 2, 5)
]
sjf(Jobs)
3. 輪轉(zhuǎn)調(diào)度 (Round Robin)
基本思想
給每個(gè)作業(yè)固定的執(zhí)行時(shí)間，根據(jù)作業(yè)到達(dá)的先后順序讓作業(yè)在固定的時(shí)間片內(nèi)執(zhí)行，執(zhí)行完成后便調(diào)度下一個(gè)進(jìn)程執(zhí)行，時(shí)間片輪轉(zhuǎn)調(diào)度不考慮進(jìn)程等待時(shí)間和執(zhí)行時(shí)間，屬于搶占式調(diào)度。優(yōu)點(diǎn)是兼顧長(zhǎng)短作業(yè)；缺點(diǎn)是平均等待時(shí)間較長(zhǎng)，上下文切換較費(fèi)時(shí)。適用于分時(shí)系統(tǒng)。
實(shí)現(xiàn)代碼（可直接運(yùn)行）
# 時(shí)間片輪轉(zhuǎn)算法
class Job:
def __init__(self, name, arrival_time, burst_time):
self.name = name
self.arrival_time = arrival_time
self.burst_time = burst_time
def round_robin(Jobs, time_slice):
print("時(shí)間片輪轉(zhuǎn)算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0
remaining_time = [job.burst_time for job in Jobs] # 用于記錄每個(gè)作業(yè)的剩余執(zhí)行時(shí)間
completed_jobs = [] # 已完成的作業(yè)隊(duì)列
total_waiting_time = 0 # 總等待時(shí)間
total_turnaround_time_rate = 0 # 總帶權(quán)周轉(zhuǎn)時(shí)間
while any(remaining > 0 for remaining in remaining_time):# 只要有作業(yè)的剩余執(zhí)行時(shí)間大于0，就繼續(xù)執(zhí)行
for i, job in enumerate(Jobs):# 遍歷作業(yè)列表
if remaining_time[i] > 0 and job not in completed_jobs: # 如果作業(yè)的剩余執(zhí)行時(shí)間大于0且作業(yè)沒(méi)有完成
if remaining_time[i] <= time_slice: # 如果作業(yè)的剩余執(zhí)行時(shí)間小于等于時(shí)間片,說(shuō)明作業(yè)可以在本輪執(zhí)行完
current_time += remaining_time[i] # 當(dāng)前時(shí)間加上作業(yè)的剩余執(zhí)行時(shí)間
remaining_time[i] = 0 # 作業(yè)的剩余執(zhí)行時(shí)間置為0
completed_jobs.append(job) # 將作業(yè)加入已完成的作業(yè)隊(duì)列
else: # 如果作業(yè)的剩余執(zhí)行時(shí)間大于時(shí)間片,說(shuō)明作業(yè)不能在本輪執(zhí)行完
current_time += time_slice # 當(dāng)前時(shí)間加上時(shí)間片
remaining_time[i] -= time_slice # 作業(yè)的剩余執(zhí)行時(shí)間減去時(shí)間片
if (len(completed_jobs)==0):
continue;
for job in completed_jobs:
completion_time = round((current_time),2) # 完成時(shí)間=當(dāng)前時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) # 周轉(zhuǎn)時(shí)間=完成時(shí)間-到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) # 等待時(shí)間=周轉(zhuǎn)時(shí)間-執(zhí)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),2) # 帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/執(zhí)行時(shí)間
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 # 累加等待時(shí)間
total_turnaround_time_rate += turnaround_time_rate # 累加帶權(quán)周轉(zhuǎn)時(shí)間
completed_jobs.remove(job)
print("\n用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
# 示例作業(yè)列表
Jobs = [
Job("P1", 0, 4),
Job("P2", 1, 3),
Job("P3", 2, 4),
Job("P4", 3, 2),
Job("P5", 4, 4)
]
time_slice = 4 # 設(shè)置時(shí)間片大小
round_robin(Jobs, time_slice)
4.優(yōu)先級(jí)調(diào)度算法
基本思想
給每個(gè)作業(yè)都設(shè)置一個(gè)優(yōu)先級(jí)，然后在調(diào)度的時(shí)候，在所有處于就緒狀態(tài)的任務(wù)中選擇優(yōu)先級(jí)最高的任務(wù)去運(yùn)行。
實(shí)現(xiàn)代碼（可直接運(yùn)行）
#非搶占式優(yōu)先級(jí)調(diào)度算法，當(dāng)所有進(jìn)程都處于就緒狀態(tài)時(shí)，按照優(yōu)先級(jí)從高到低順序選擇一個(gè)進(jìn)程執(zhí)行
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 #作業(yè)優(yōu)先級(jí)
def priority_scheduling(Jobs):
print("優(yōu)先級(jí)調(diào)度算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t優(yōu)先級(jí)\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0
completed_jobs = [] #記錄已經(jīng)完成的作業(yè)列表
while Jobs:
max_priority = float('-inf') #設(shè)置負(fù)無(wú)窮小為目前的最大優(yōu)先級(jí)
selected_job = None #記錄當(dāng)前選中的作業(yè)
for job in Jobs:#選出到達(dá)作業(yè)中的優(yōu)先級(jí)最高的作業(yè)
if job.arrival_time <= current_time and job not in completed_jobs: #如果作業(yè)到達(dá)時(shí)間小于(早于)等于當(dāng)前時(shí)間，并且作業(yè)還沒(méi)有被完成
if job.priority > max_priority:#如果作業(yè)優(yōu)先級(jí)大于當(dāng)前最大優(yōu)先級(jí)，則將當(dāng)前作業(yè)設(shè)為選中作業(yè)
max_priority = job.priority
selected_job = job
if selected_job is None: #如果沒(méi)有選中作業(yè)，則說(shuō)明當(dāng)前時(shí)間沒(méi)有到達(dá)作業(yè)，則直接跳過(guò)
current_time += 1
continue
completion_time = round((current_time + selected_job.burst_time),2) # 完成時(shí)間 = 當(dāng)前時(shí)間 + 作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - selected_job.arrival_time),2)#周轉(zhuǎn)時(shí)間 = 完成時(shí)間 - 到達(dá)時(shí)間
waiting_time = round((turnaround_time - selected_job.burst_time),2)#等待時(shí)間 = 周轉(zhuǎn)時(shí)間 - 運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / selected_job.burst_time),2) #帶權(quán)周轉(zhuǎn)時(shí)間 = 周轉(zhuǎn)時(shí)間 / 運(yùn)行時(shí)間
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 #當(dāng)前時(shí)間更新為完成時(shí)間
completed_jobs.append(selected_job) #將選中作業(yè)加入已完成作業(yè)列表
Jobs.remove(selected_job) #將選中作業(yè)從作業(yè)列表中移除
total_waiting_time = sum(turnaround_time - job.burst_time for job in completed_jobs)#計(jì)算總等待時(shí)間
total_turnaround_time_rate = sum(turnaround_time / job.burst_time for job in completed_jobs)#計(jì)算總帶權(quán)周轉(zhuǎn)時(shí)間
print("\n用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(completed_jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
# 示例作業(yè)列表
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.高響應(yīng)比優(yōu)先調(diào)度算法（HRRN）
基本思想
高響應(yīng)比優(yōu)先調(diào)度算法（Highest?Response?Ratio?Next）是一種介于FCFS（先來(lái)先服務(wù)算法）與SJF（短作業(yè)優(yōu)先算法）之間的折中算法，根據(jù)作業(yè)的響應(yīng)比驚醒調(diào)度。既考慮作業(yè)等待時(shí)間又考慮作業(yè)運(yùn)行時(shí)間，既照顧短作業(yè)又不使長(zhǎng)作業(yè)等待時(shí)間過(guò)長(zhǎng)，改進(jìn)了調(diào)度性能。
響應(yīng)比=(作業(yè)等待時(shí)間+作業(yè)運(yùn)行時(shí)間)/作業(yè)運(yùn)行時(shí)間
實(shí)現(xiàn)代碼
#響應(yīng)比=（等待時(shí)間+作業(yè)運(yùn)行時(shí)間）/作業(yè)運(yùn)行時(shí)間
#響應(yīng)比越大，優(yōu)先級(jí)越高
class Job:
def __init__(self, name, arrival_time, burst_time):
self.name = name
self.arrival_time = arrival_time
self.burst_time = burst_time
#計(jì)算響應(yīng)比
def calculate_response_ratio(job, current_time):
wait_time = max(0, current_time - job.arrival_time) #更新等待時(shí)間
response_ratio = (wait_time + job.burst_time) / job.burst_time #計(jì)算響應(yīng)比
return response_ratio
def hrrn(Jobs):
print("高響應(yīng)比優(yōu)先算法（HRRN）-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0 #當(dāng)前時(shí)間
completed_jobs = [] #已完成的作業(yè)隊(duì)列
while Jobs:
max_response_ratio = -1 #設(shè)置目前最大響應(yīng)比為-1，響應(yīng)比越大，優(yōu)先級(jí)越高
selected_job = None
for job in Jobs:#計(jì)算每個(gè)作業(yè)的當(dāng)前響應(yīng)比
if job.arrival_time <= current_time and job not in completed_jobs: #如果作業(yè)已到達(dá)且未完成
response_ratio = calculate_response_ratio(job, current_time) #計(jì)算響應(yīng)比
if response_ratio > max_response_ratio:
max_response_ratio = response_ratio
selected_job = job
if selected_job is None: #如果沒(méi)有作業(yè)被選中,說(shuō)明當(dāng)前時(shí)間沒(méi)有作業(yè)到達(dá)
current_time += 1 #當(dāng)前時(shí)間加1
continue
completion_time = round((current_time + selected_job.burst_time),2) #計(jì)算完成時(shí)間
turnaround_time = round((completion_time - selected_job.arrival_time),2) #計(jì)算周轉(zhuǎn)時(shí)間
waiting_time = round((turnaround_time - selected_job.burst_time),2) #計(jì)算等待時(shí)間
turnaround_time_rate =round((turnaround_time / selected_job.burst_time),2) #計(jì)算帶權(quán)周轉(zhuǎn)時(shí)間
print(
f"{selected_job.name}\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 #更新當(dāng)前時(shí)間
completed_jobs.append(selected_job) #將已完成的作業(yè)加入已完成作業(yè)隊(duì)列
Jobs.remove(selected_job) #將已完成的作業(yè)從作業(yè)隊(duì)列中移除
total_waiting_time = sum(turnaround_time - job.burst_time for job in completed_jobs) #計(jì)算總等待時(shí)間
total_turnaround_time_rate = sum(turnaround_time / job.burst_time for job in completed_jobs) #計(jì)算總帶權(quán)周轉(zhuǎn)時(shí)間
print("\n用時(shí):", current_time)
print("平均等待時(shí)間:", round(total_waiting_time / len(completed_jobs)),2)
avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", round(avg_turnaround_time_rate),2)
# 示例作業(yè)列表
Jobs = [
Job("P1", 8, 2),
Job("P2", 8.3, 0.5),
Job("P3", 8.5, 0.1),
Job("P4", 9, 0.4),
#Job("P5", 4, 4)
]
hrrn(Jobs)
6.整合
這里我將五個(gè)算法封裝在了同一個(gè)py文件中，方便用多個(gè)測(cè)試用例調(diào)用這些算法
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 = priority
def fcfs(Jobs):
print("作業(yè)-到達(dá)時(shí)間-服務(wù)時(shí)間-優(yōu)先權(quán)")
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("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
for job in Jobs:
if current_time < job.arrival_time:
current_time = job.arrival_time
completion_time = round((current_time + job.burst_time),2) # 作業(yè)完成時(shí)間=當(dāng)前時(shí)間+作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) # 作業(yè)周轉(zhuǎn)時(shí)間=完成時(shí)間-到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) # 等待時(shí)間=周轉(zhuǎn)時(shí)間-運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),) # 帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/運(yùn)行時(shí)間
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用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
def sjf(Jobs):
print("最短作業(yè)優(yōu)先算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0 # 當(dāng)前時(shí)間，記錄每次作業(yè)運(yùn)行完后的時(shí)間點(diǎn)
total_waiting_time = 0 # 總等待時(shí)間
total_turnaround_time_rate = 0 # 總帶權(quán)周轉(zhuǎn)時(shí)間
Jobs.sort(key=lambda x: (x.burst_time, x.arrival_time)) # 按到達(dá)時(shí)間和運(yùn)行時(shí)間排序
#for job in Jobs:
# print(job.name)
#這里的for循環(huán)，判斷是否有0時(shí)刻到達(dá)的作業(yè)，將0時(shí)刻到達(dá)的執(zhí)行時(shí)間最短的作業(yè)放到隊(duì)列首部先執(zhí)行
for job in Jobs:
if job.arrival_time==0:
Jobs.remove(job)
Jobs.insert(0,job) #插入到隊(duì)首
break #上面已經(jīng)按照運(yùn)行時(shí)間拍好序了，所以這里找到的作業(yè)是所有0時(shí)刻到達(dá)的作業(yè)中執(zhí)行時(shí)間最短的，直接退出循環(huán)
for job in Jobs:
if current_time < job.arrival_time: # 當(dāng)前時(shí)間小于作業(yè)到達(dá)時(shí)間，則等待
current_time = job.arrival_time # 當(dāng)前時(shí)間更新為作業(yè)到達(dá)時(shí)間
completion_time = round((current_time + job.burst_time),2) # 作業(yè)完成時(shí)間=當(dāng)前時(shí)間+作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) #周轉(zhuǎn)時(shí)間=作業(yè)完成時(shí)間-作業(yè)到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) #等待時(shí)間=周轉(zhuǎn)時(shí)間-作業(yè)運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),2) #帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/作業(yè)運(yùn)行時(shí)間
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 # 當(dāng)前時(shí)間更新為作業(yè)完成時(shí)間
total_waiting_time += waiting_time # 總等待時(shí)間累加
total_turnaround_time_rate += turnaround_time_rate # 總帶權(quán)周轉(zhuǎn)時(shí)間累加
print("\n作業(yè)總用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
def rr(Jobs, time_slice):
print("時(shí)間片輪轉(zhuǎn)算法-----------------")
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
current_time = 0
remaining_time = [job.burst_time for job in Jobs] # 用于記錄每個(gè)作業(yè)的剩余執(zhí)行時(shí)間
completed_jobs = [] # 已完成的作業(yè)隊(duì)列
total_waiting_time = 0 # 總等待時(shí)間
total_turnaround_time_rate = 0 # 總帶權(quán)周轉(zhuǎn)時(shí)間
while any(remaining > 0 for remaining in remaining_time):# 只要有作業(yè)的剩余執(zhí)行時(shí)間大于0，就繼續(xù)執(zhí)行
for i, job in enumerate(Jobs):# 遍歷作業(yè)列表
if remaining_time[i] > 0 and job not in completed_jobs: # 如果作業(yè)的剩余執(zhí)行時(shí)間大于0且作業(yè)沒(méi)有完成
if remaining_time[i] <= time_slice: # 如果作業(yè)的剩余執(zhí)行時(shí)間小于等于時(shí)間片,說(shuō)明作業(yè)可以在本輪執(zhí)行完
current_time += remaining_time[i] # 當(dāng)前時(shí)間加上作業(yè)的剩余執(zhí)行時(shí)間
remaining_time[i] = 0 # 作業(yè)的剩余執(zhí)行時(shí)間置為0
completed_jobs.append(job) # 將作業(yè)加入已完成的作業(yè)隊(duì)列
else: # 如果作業(yè)的剩余執(zhí)行時(shí)間大于時(shí)間片,說(shuō)明作業(yè)不能在本輪執(zhí)行完
current_time += time_slice # 當(dāng)前時(shí)間加上時(shí)間片
remaining_time[i] -= time_slice # 作業(yè)的剩余執(zhí)行時(shí)間減去時(shí)間片
if (len(completed_jobs)==0):
continue;
for job in completed_jobs:
completion_time = round((current_time),2) # 完成時(shí)間=當(dāng)前時(shí)間
turnaround_time = round((completion_time - job.arrival_time),2) # 周轉(zhuǎn)時(shí)間=完成時(shí)間-到達(dá)時(shí)間
waiting_time = round((turnaround_time - job.burst_time),2) # 等待時(shí)間=周轉(zhuǎn)時(shí)間-執(zhí)行時(shí)間
turnaround_time_rate = round((turnaround_time / job.burst_time),2) # 帶權(quán)周轉(zhuǎn)時(shí)間=周轉(zhuǎn)時(shí)間/執(zhí)行時(shí)間
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 # 累加等待時(shí)間
total_turnaround_time_rate += turnaround_time_rate # 累加帶權(quán)周轉(zhuǎn)時(shí)間
completed_jobs.remove(job)
print("\n用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(Jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(Jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
def priority_scheduling(Jobs):
print("作業(yè)-到達(dá)時(shí)間-服務(wù)時(shí)間-優(yōu)先權(quán)")
for Job in Jobs:
if(Job.priority==None):
Job.priority=0
print(f"{Job.name}\t{Job.arrival_time}\t{Job.burst_time}\t{Job.priority}")
current_time = 0
completed_jobs = []
print("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t優(yōu)先級(jí)\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
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) # 完成時(shí)間 = 當(dāng)前時(shí)間 + 作業(yè)運(yùn)行時(shí)間
turnaround_time = round((completion_time - selected_job.arrival_time),2)#周轉(zhuǎn)時(shí)間 = 完成時(shí)間 - 到達(dá)時(shí)間
waiting_time = round((turnaround_time - selected_job.burst_time),2)#等待時(shí)間 = 周轉(zhuǎn)時(shí)間 - 運(yùn)行時(shí)間
turnaround_time_rate = round((turnaround_time / selected_job.burst_time),2) #帶權(quán)周轉(zhuǎn)時(shí)間 = 周轉(zhuǎn)時(shí)間 / 運(yùn)行時(shí)間
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用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(completed_jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
def hrrn(Jobs):
print("作業(yè)-到達(dá)時(shí)間-服務(wù)時(shí)間-優(yōu)先權(quán)")
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("進(jìn)程\t到達(dá)時(shí)間\t執(zhí)行時(shí)間\t完成時(shí)刻\t周轉(zhuǎn)時(shí)間\t等待時(shí)間\t帶權(quán)周轉(zhuǎn)時(shí)間")
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) #計(jì)算完成時(shí)間
turnaround_time = round((completion_time - selected_job.arrival_time),2) #計(jì)算周轉(zhuǎn)時(shí)間
waiting_time = round((turnaround_time - selected_job.burst_time),2) #計(jì)算等待時(shí)間
turnaround_time_rate =round((turnaround_time / selected_job.burst_time),2) #計(jì)算帶權(quán)周轉(zhuǎn)時(shí)間
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用時(shí):", current_time)
print("平均等待時(shí)間:", total_waiting_time / len(completed_jobs))
avg_turnaround_time_rate = total_turnaround_time_rate / len(completed_jobs)
print("平均帶權(quán)周轉(zhuǎn)時(shí)間:", avg_turnaround_time_rate)
# 作業(yè)列表
Jobs2 = [
Job("P1", 0, 9),
Job("P2", 0, 6),
Job("P3", 0, 3),
Job("P4", 0, 5),
Job("P5", 0, 9)
]
Jobs = [
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("Complete by 黃奔o(jì)n Nov. 1 at Jiangxi University of Traditional Chinese Medicine")
print("先來(lái)先服務(wù)算法-------------------------------------------------------------------------------------------------------------------------------")
fcfs(Jobs.copy())#淺拷貝，防止原列表被修改
print("\n最短作業(yè)優(yōu)先算法-------------------------------------------------------------------------------------------------------------------------------")
sjf(Jobs.copy())
print("\n時(shí)間片輪轉(zhuǎn)算法-------------------------------------------------------------------------------------------------------------------------------")
time_slice = 3
rr(Jobs.copy(), time_slice)
print("\n優(yōu)先級(jí)調(diào)度算法-------------------------------------------------------------------------------------------------------------------------------")
priority_scheduling(Jobs.copy())
print("\n高響應(yīng)比優(yōu)先算法-----------------------------------------------------------------------------------------------------------------------------")
hrrn(Jobs.copy())
運(yùn)行結(jié)果
總結(jié)
通過(guò)一天的努力，我成功地編寫(xiě)了用代碼模擬進(jìn)程調(diào)度算法的示例，包括先來(lái)先服務(wù)、最短作業(yè)優(yōu)先、輪轉(zhuǎn)調(diào)度和優(yōu)先級(jí)調(diào)度算法。雖然這個(gè)實(shí)踐實(shí)現(xiàn)的東西很簡(jiǎn)單，還有很多實(shí)際問(wèn)題、特殊情況沒(méi)有考慮到，但是這個(gè)過(guò)程不僅加深了我對(duì)操作系統(tǒng)進(jìn)程調(diào)度算法的理解，還讓我更深入地體驗(yàn)了算法在實(shí)際應(yīng)用中的工作原理。我希望這種實(shí)踐能夠幫助我、同時(shí)幫助你們更好地掌握這一重要概念。文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-734698.html
寫(xiě)這篇文章是為了記錄我今天的成果，當(dāng)然以后我肯定用得上這些代碼，畢竟我還沒(méi)學(xué)操作系統(tǒng)這門(mén)課，下學(xué)期才學(xué)，我相信肯定有要我們算什么“周轉(zhuǎn)時(shí)間”、“平均等待時(shí)間”的這些題目，大伙如果碰到了這種題目，可以用代碼來(lái)檢測(cè)一下自己的結(jié)果是不是正確（當(dāng)然我寫(xiě)的示例過(guò)于簡(jiǎn)單，實(shí)際的調(diào)度比這個(gè)復(fù)雜了多，有些特殊情況可能沒(méi)有考慮到）哈哈哈，如果你覺(jué)得我說(shuō)的沒(méi)錯(cuò)就給我點(diǎn)個(gè)贊唄！文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-734698.html
到了這里，關(guān)于用代碼模擬操作系統(tǒng)進(jìn)程調(diào)度算法（Python）的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
