多道技術與程序
阿新 • • 發佈：2021-10-18
目錄
多道技術與程序
1 多道技術
單核實現併發的效果
1.1 必備知識點
-
併發
看起來像同時執行的就可以稱之為併發
-
並行
真正意義上的同時執行
ps:
- 並行肯定算併發
- 單核的計算機肯定不能實現並行，但是可以實現併發！！！
補充：我們直接假設單核就是一個核，幹活的就一個人，不要考慮cpu裡面的核心數
1.2 多道技術圖解
節省多個程式執行的總耗時
1.3 多道技術重點知識
空間上的服用與時間上的服用
-
空間上的複用
多個程式公用一套計算機硬體
-
時間上的複用
例子:洗衣服30s，做飯50s，燒水30s
單道需要110s，多道只需要任務做長的那一個 切換節省時間
例子:邊吃飯邊玩遊戲 儲存狀態
切換+儲存狀態
"""
切換(CPU)分為兩種情況
1.當一個程式遇到IO操作的時候，作業系統會剝奪該程式的CPU執行許可權
作用:提高了CPU的利用率 並且也不影響程式的執行效率
2.當一個程式長時間佔用CPU的時候，操作吸引也會剝奪該程式的CPU執行許可權
弊端:降低了程式的執行效率(原本時間+切換時間)
""" 2 程序理論
2.1 必備知識點
程式與程序的區別
"""
程式就是一堆躺在硬碟上的程式碼，是“死”的
程序則表示程式正在執行的過程，是“活”的
"""
2.2 程序排程
-
先來先服務排程演算法
"""對長作業有利，對短作業無益""" -
短作業優先排程演算法
"""對短作業有利，多長作業無益""" -
時間片輪轉法+多級反饋佇列
2.3 程序執行的三狀態圖
2.4 重要概念
-
同步和非同步
"""描述的是任務的提交方式""" 同步:任務提交之後，原地等待任務的返回結果，等待的過程中不做任何事(乾等) 程式層面上表現出來的感覺就是卡住了 非同步:任務提交之後，不原地等待任務的返回結果，直接去做其他事情 我提交的任務結果如何獲取？ 任務的返回結果會有一個非同步回撥機制自動處理 -
阻塞非阻塞
"""描述的程式的執行狀態""" 阻塞:阻塞態 非阻塞:就緒態、執行態 理想狀態:我們應該讓我們的寫的程式碼永遠處於就緒態和執行態之間切換
上述概念的組合:最高效的一種組合就是非同步非阻塞
3 程序實踐
3.1 開啟程序的兩種方式
# 第一種方式 Process
from multiprocessing import Process
import time
def task(name):
print('%s is running'%name)
time.sleep(3)
print('%s is over'%name)
if __name__ == '__main__':
# 1 建立一個物件
p = Process(target=task, args=('jason',))
# 容器型別哪怕裡面只有1個元素 建議要用逗號隔開
# 2 開啟程序
p.start() # 告訴作業系統幫你建立一個程序 非同步
print('主')
# 第二種方式 類的繼承
from multiprocessing import Process
import time
class MyProcess(Process):
def run(self):
print('hello bf girl')
time.sleep(1)
print('get out!')
if __name__ == '__main__':
p = MyProcess()
p.start()
print('主')
3.2 join方法
join是讓主程序等待子程序程式碼執行結束之後，再繼續執行。不影響其他子程序的執行
from multiprocessing import Process
import time
def task(name, n):
print('%s is running'%name)
time.sleep(n)
print('%s is over'%name)
if __name__ == '__main__':
# p1 = Process(target=task, args=('jason', 1))
# p2 = Process(target=task, args=('egon', 2))
# p3 = Process(target=task, args=('tank', 3))
# start_time = time.time()
# p1.start()
# p2.start()
# p3.start() # 僅僅是告訴作業系統要建立程序
# # time.sleep(50000000000000000000)
# # p.join() # 主程序等待子程序p執行結束之後再繼續往後執行
# p1.join()
# p2.join()
# p3.join()
start_time = time.time()
p_list = []
for i in range(1, 4):
p = Process(target=task, args=('子程序%s'%i, i))
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('主', time.time() - start_time)
3.3 程序之間資料相互隔離
from multiprocessing import Process
money = 100
def task():
global money # 區域性修改全域性
money = 666
print('子',money)
if __name__ == '__main__':
p = Process(target=task)
p.start()
p.join()
print(money)
3.4 總結
"""
建立程序就是在記憶體中申請一塊記憶體空間將需要執行的程式碼丟進去
一個程序對應在記憶體中就是一塊獨立的記憶體空間
多個程序對應在記憶體中就是多塊獨立的記憶體空間
程序與程序之間資料預設情況下是無法直接互動,如果想互動可以藉助於第三方工具、模組
"""
3.5 程序物件及其他方法
"""
一臺計算機上面執行著很多程序，那麼計算機是如何區分並管理這些程序服務端的呢？
計算機會給每一個執行的程序分配一個PID號
如何檢視
windows電腦
進入cmd輸入tasklist即可檢視
tasklist |findstr PID檢視具體的程序
mac電腦
進入終端之後輸入ps aux
ps aux|grep PID檢視具體的程序
"""
from multiprocessing import Process, current_process
current_process().pid # 檢視當前程序的程序號
import os
os.getpid() # 檢視當前程序程序號
os.getppid() # 檢視當前程序的父程序程序號
p.terminate() # 殺死當前程序
# 是告訴作業系統幫你去殺死當前程序 但是需要一定的時間 而程式碼的執行速度極快
time.sleep(0.1)
print(p.is_alive()) # 判斷當前程序是否存活
3.6 殭屍程序與孤兒程序
# 殭屍程序
"""
死了但是沒有死透
當你開設了子程序之後 該程序死後不會立刻釋放佔用的程序號
因為我要讓父程序能夠檢視到它開設的子程序的一些基本資訊 佔用的pid號 執行時間。。。
所有的程序都會步入殭屍程序
父程序不死並且在無限制的建立子程序並且子程序也不結束
回收子程序佔用的pid號
父程序等待子程序執行結束
父程序呼叫join方法
"""
# 孤兒程序
"""
子程序存活，父程序意外死亡
作業系統會開設一個“兒童福利院”專門管理孤兒程序回收相關資源
"""
3.7 守護程序
守護程序是一種程序,它會在它的主程序結束後也立即結束
from multiprocessing import Process
import time
def task(name):
print('%s總管正在活著'% name)
time.sleep(3)
print('%s總管正在死亡' % name)
if __name__ == '__main__':
p = Process(target=task,args=('egon',))
# p = Process(target=task,kwargs={'name':'egon'})
p.daemon = True # 將程序p設定成守護程序 這一句一定要放在start方法上面才有效否則會直接報錯
p.start()
print('皇帝jason壽終正寢')
3.8 互斥鎖
因為多個程序之間資料是隔離的,所以多個程序操作同一份資料的時候，會出現資料錯亂的問題
針對上述問題，解決方式就是加鎖處理:將併發變成序列，雖然犧牲效率但是保證了資料的安全
from multiprocessing import Process, Lock
import json
import time
import random
# 查票
def search(i):
# 檔案操作讀取票數
with open('data','r',encoding='utf8') as f:
dic = json.load(f)
print('使用者%s查詢餘票：%s'%(i, dic.get('ticket_num')))
# 字典取值不要用[]的形式 推薦使用get 你寫的程式碼打死都不能報錯！！！
# 買票 1.先查 2.再買
def buy(i):
# 先查票
with open('data','r',encoding='utf8') as f:
dic = json.load(f)
# 模擬網路延遲
time.sleep(random.randint(1,3))
# 判斷當前是否有票
if dic.get('ticket_num') > 0:
# 修改資料庫 買票
dic['ticket_num'] -= 1
# 寫入資料庫
with open('data','w',encoding='utf8') as f:
json.dump(dic,f)
print('使用者%s買票成功'%i)
else:
print('使用者%s買票失敗'%i)
# 整合上面兩個函式
def run(i, mutex):
search(i)
# 給買票環節加鎖處理
# 搶鎖
mutex.acquire()
buy(i)
# 釋放鎖
mutex.release()
if __name__ == '__main__':
# 在主程序中生成一把鎖 讓所有的子程序搶 誰先搶到誰先買票
mutex = Lock()
for i in range(1,11):
p = Process(target=run, args=(i, mutex))
p.start()
"""
擴充套件 行鎖 表鎖
注意：
1.鎖不要輕易的使用，容易造成死鎖現象(我們寫程式碼一般不會用到，都是內部封裝好的)
2.鎖只在處理資料的部分加來保證資料安全(只在爭搶資料的環節加鎖處理即可)
"""
3.9 程序間通訊
3.9.1 佇列Queue模組
管道:subprocess
佇列:管道+鎖
"""
佇列:先進先出
堆疊:先進後出
"""
from multiprocessing import Queue
# 建立一個佇列
q = Queue(5) # 括號內可以傳數字 標示生成的佇列最大可以同時存放的資料量
# 往佇列中存資料
q.put(111)
q.put(222)
q.put(333)
# print(q.full()) # 判斷當前佇列是否滿了
# print(q.empty()) # 判斷當前佇列是否空了
q.put(444)
q.put(555)
# print(q.full()) # 判斷當前佇列是否滿了
# q.put(666) # 當佇列資料放滿了之後 如果還有資料要放程式會阻塞 直到有位置讓出來 不會報錯
"""
存取資料 存是為了更好的取
千方百計的存、簡單快捷的取
同在一個屋簷下
差距為何那麼大
"""
# 去佇列中取資料
v1 = q.get()
v2 = q.get()
v3 = q.get()
v4 = q.get()
v5 = q.get()
# print(q.empty())
# V6 = q.get_nowait() # 沒有資料直接報錯queue.Empty
# v6 = q.get(timeout=3) # 沒有資料之後原地等待三秒之後再報錯 queue.Empty
try:
v6 = q.get(timeout=3)
print(v6)
except Exception as e:
print('一滴都沒有了!')
# # v6 = q.get() # 佇列中如果已經沒有資料的話 get方法會原地阻塞
# print(v1, v2, v3, v4, v5, v6)
"""
q.full()
q.empty()
q.get_nowait()
在多程序的情況下是不精確
"""
3.9.2 IPC機制
通過佇列進行程序間的通訊（Inter-Process Communication，程序間通訊）
IPC方法包括管道（PIPE）、訊息排隊、旗語、共用記憶體以及套接字（Socket）
from multiprocessing import Queue, Process
"""
研究思路
1.主程序跟子程序藉助於佇列通訊
2.子程序跟子程序藉助於佇列通訊
"""
def producer(q):
q.put('我是23號技師 很高興為您服務')
def consumer(q):
print(q.get())
if __name__ == '__main__':
q = Queue()
p = Process(target=producer,args=(q,))
p1 = Process(target=consumer,args=(q,))
p.start()
p1.start()
3.10 生產者消費者模型
"""
生產者:生產/製造東西的
消費者:消費/處理東西的
該模型除了上述兩個之外還需要一個媒介
生活中的例子做包子的將包子做好後放在蒸籠(媒介)裡面，買包子的取蒸籠裡面拿
廚師做菜做完之後用盤子裝著給你消費者端過去
生產者和消費者之間不是直接做互動的，而是藉助於媒介做互動
生產者(做包子的) + 訊息佇列(蒸籠) + 消費者(吃包子的)
"""
from multiprocessing import Process, Queue, JoinableQueue
import time
import random
def producer(name,food,q):
for i in range(5):
data = '%s生產了%s%s'%(name,food,i)
# 模擬延遲
time.sleep(random.randint(1,3))
print(data)
# 將資料放入 佇列中
q.put(data)
def consumer(name,q):
# 消費者胃口很大 光碟行動
while True:
food = q.get() # 沒有資料就會卡住
# 判斷當前是否有結束的標識
# if food is None:break
time.sleep(random.randint(1,3))
print('%s吃了%s'%(name,food))
q.task_done() # 告訴佇列你已經從裡面取出了一個數據並且處理完畢了
if __name__ == '__main__':
# q = Queue()
q = JoinableQueue()
p1 = Process(target=producer,args=('大廚egon','包子',q))
p2 = Process(target=producer,args=('馬叉蟲tank','泔水',q))
c1 = Process(target=consumer,args=('春哥',q))
c2 = Process(target=consumer,args=('新哥',q))
p1.start()
p2.start()
# 將消費者設定成守護程序
c1.daemon = True
c2.daemon = True
c1.start()
c2.start()
p1.join()
p2.join()
# 等待生產者生產完畢之後 往佇列中新增特定的結束符號
# q.put(None) # 肯定在所有生產者生產的資料的末尾
# q.put(None) # 肯定在所有生產者生產的資料的末尾
q.join() # 等待佇列中所有的資料被取完再執行往下執行程式碼
"""
JoinableQueue 每當你往該佇列中存入資料的時候 內部會有一個計數器+1
沒當你呼叫task_done的時候 計數器-1
q.join() 當計數器為0的時候 才往後執行
"""
# 只要q.join執行完畢 說明消費者已經處理完資料了 消費者就沒有存在的必要了
# 所以可以將消費者設定成守護程序