4.18 程序排程模擬演算法
阿新 • • 發佈：2022-04-18
執行緒詳解
- 內建佇列
- IPC機制(程序間通訊)
- 生產者消費者模型
- 執行緒理論(重要)
- 開設執行緒的兩種方式
- 執行緒實現TCP服務端併發
- 執行緒join方法
- 執行緒間資料共享
- 守護執行緒
- GIL全域性直譯器鎖
內建佇列
佇列：先進先出(使用頻率高)
堆疊：先進後出
以後我們會直接使用別人封裝好的訊息佇列 實現各種資料傳輸
from multiprocessing import Queue # 自定義佇列的長度 q = Queue(5) # 朝佇列中存放資料 q.put(111) q.put(222) q.put(333) print(q.full()) # False 判斷佇列是否滿了 q.put(444) q.put(555) print(q.full()) # True # q.put(666) # 超出最大長度 原地阻塞等待佇列中出現空位 print(q.get()) print(q.get()) print(q.empty()) # False 判斷佇列是否空了 print(q.get()) print(q.get()) print(q.get()) print(q.empty()) # True # print(q.get()) # 佇列中沒有值 繼續獲取則阻塞等待佇列中給值 print(q.get_nowait()) # 佇列中如果沒有值 直接報錯 """ full() empty() get_nowait() 上述方法能否在併發的場景下精準使用??? 不能用!!! 之所以介紹佇列是因為它可以支援程序間資料通訊 """
超出最大長度 原地阻塞等待佇列中出現空位
IPC機制(程序間通訊)
1.主程序與子程序資料互動
2.兩個子程序資料互動
本質:不同記憶體空間中的程序資料互動
from multiprocessing import Process, Queue def producer(q): # print('子程序producer從佇列中取值>>>:', q.get()) q.put('子程序producer往佇列中新增值') def consumer(q): print('子程序consumer從佇列中取值>>>:', q.get()) if __name__ == '__main__': q = Queue() p = Process(target=producer, args=(q, )) p1 = Process(target=consumer, args=(q,)) p.start() p1.start() # 主程序往佇列中存放資料123 # q.put(123) print('主程序'
如果解開q.put(123)，那麼取值會顯示123，因為put(123)處於主執行緒中，比子程序執行快
生產者消費者模型
生產者：負責生產/製作資料
消費者：負責消費/處理資料
比如在爬蟲領域中
會先通過程式碼爬取網頁資料(爬取網頁的程式碼就可以稱之為是生產者)
之後針對網頁資料做篩選處理(處理網頁的程式碼就可以稱之為消費者)
如果使用程序來演示
除了有至少兩個程序之外 還需要一個媒介(訊息佇列)
以後遇到該模型需要考慮的問題其實就是供需平衡的問題
生產力與消費力要均衡
程式碼示例：
from multiprocessing import Process, JoinableQueue import time import random def producer(name, food, q): for i in range(5): data = f'{name}生產了{food}{i}' print(data) time.sleep(random.randint(1, 3)) # 模擬產生過程 q.put(data) def consumer(name, q): while True: food = q.get() time.sleep(random.random()) print(f'{name}吃了{food}') q.task_done() # 每次去完資料必須給佇列一個反饋 if __name__ == '__main__': q = JoinableQueue() p1 = Process(target=producer, args=('大廚jason', '韭菜炒蛋', q)) p2 = Process(target=producer, args=('老闆kevin', '祕製小漢堡', q)) c1 = Process(target=consumer, args=('濤濤', q)) c2 = Process(target=consumer, args=('龍龍', q)) c1.daemon = True c2.daemon = True p1.start() p2.start() c1.start() c2.start() # 生產者生產完所有資料之後 往佇列中新增結束的訊號 p1.join() p2.join() # q.put(None) # 結束訊號的個數要跟消費者個數一致才可以 # q.put(None) """佇列中其實已經自己加了鎖 所以多程序取值也不會衝突 並且取走了就沒了""" q.join() # 等待佇列中資料全部被取出(一定要讓生產者全部結束才能判斷正確) """執行完上述的join方法表示消費者也已經消費完資料了""" print('主執行緒')
執行緒理論
1.什麼是執行緒？
程序：資源單位
執行緒：執行單位
程序相當於車間(一個個空間)，執行緒相當於車間裡面的流水線(真正幹 活的)
一個程序中至少有一個執行緒
程序僅僅是在記憶體中開闢一塊空間(提供執行緒工作所需的資源), 執行緒真 正被CPU執行，執行緒需要的資源跟所在程序的要
2.為什麼要有執行緒?
開設執行緒的消耗遠遠小於程序
開程序:
1.申請記憶體空間
2.拷貝程式碼
開執行緒
一個程序內可以開設多個執行緒 無需申請記憶體空間、拷貝程式碼
一個程序內的多個執行緒資料是共享的
開設執行緒的兩種方式
eg:
1.函式建立：
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(3)
print(f'{name} is over')
# 建立執行緒無需在__main__下面編寫 但是為了統一 還是習慣在子程式碼中寫
t = Thread(target=task, args=('jason', ))
t.start() # 建立執行緒的開銷極小 幾乎是一瞬間就可以建立
print('主執行緒')
2.通過類建立：
class MyThread(Thread):
def __init__(self, username):
super().__init__()
self.username = username
def run(self):
print(f'{self.username} jason is running')
time.sleep(3)
print(f'{self.username} is over')
t = MyThread('jasonNB')
t.start()
print('主執行緒')
建立執行緒無需在main下面編寫 但是為了統一 還是習慣在子程式碼中寫
建立執行緒的開銷極小 幾乎是一瞬間就可以建立
執行緒實現TCP服務端的併發
服務端：
import socket
import time
from threading import Thread
server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen()
def talk(sock):
while True:
data = sock.recv(1024)
print(data.decode('utf8'))
sock.send(data.upper())
while True:
sock, addr = server.accept()
# 每來一個客戶端就建立一個執行緒做資料互動
t = Thread(target=talk, args=(sock,))
t.start()
客戶端：
import socket
import time
client = socket.socket()
client.connect(('127.0.0.1', 8080))
while True:
client.send(b'hello baby')
# time.sleep(3)
data = client.recv(1024)
print(data.decode('utf8'))
執行緒join方法
主執行緒為什麼要等著子執行緒結束才會結束整個程序
因為主執行緒結束也就標誌著整個程序的結束 要確保子執行緒執行過程中所需的各項資源
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(3)
print(f'{name} is over')
t = Thread(target=task, args=('jason', ))
t.start()
t.join() # 主執行緒程式碼等待子執行緒程式碼執行完畢之後再往下執行
print('主執行緒')
同一個程序內的多個執行緒資料共享
from threading import Thread
money = 10000000000
def task():
global money
money = 1
t = Thread(target=task)
t.start()
t.join()
print(money)
執行緒物件屬性和方法
1.驗證一個程序下的多個執行緒是否真的處於一個程序
驗證確實如此
2.統計程序下活躍的執行緒數
active_count() # 注意主執行緒也算!!!
3.獲取執行緒的名字
1.current_thread().name
MainThread 主執行緒
Thread-1、Thread-2 子執行緒
2.self.name
程式碼示例：
def task():
time.sleep(1)
print('子執行緒獲取程序號>>>:',os.getpid())
print(current_thread().name)
t = Thread(target=task)
t1 = Thread(target=task)
t.start()
t1.start()
print(active_count())
print(current_thread().name)
print('主執行緒獲取程序號>>>:', os.getpid())
class MyThread(Thread):
# def __init__(self, name):
# super().__init__()
# self.name = name
def run(self):
print(self.name)
t1 = MyThread()
t2 = MyThread()
t1.start()
t2.start()
守護執行緒
原理：主執行緒要等待所有非守護執行緒結束才可以結束
from threading import Thread
import time
def task(name):
print(f'{name} is running')
time.sleep(3)
print(f'{name} is over')
t1 = Thread(target=task, args=('jason',))
t2 = Thread(target=task, args=('kevin',))
t1.daemon = True
t1.start()
# t2.start()
print('主執行緒')
如果把t2.start()解開，那麼才會體現守護執行緒它的原理
GIL全域性直譯器鎖
# 官方文件
In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly
because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)
"""
1.回顧
python直譯器的類別有很多
Cpython Jpython Ppython
垃圾回收機制
應用計數、標記清除、分代回收
GIL只存在於CPython直譯器中，不是python的特徵
GIL是一把互斥鎖用於阻止同一個程序下的多個執行緒同時執行
原因是因為CPython直譯器中的垃圾回收機制不是執行緒安全的
反向驗證GIL的存在 如果不存在會產生垃圾回收機制與正常執行緒之間資料錯亂
GIL是加在CPython直譯器上面的互斥鎖
同一個程序下的多個執行緒要想執行必須先搶GIL鎖 所以同一個程序下多個執行緒肯定不能同時執行 即無法利用多核優勢
強調:同一個程序下的多個執行緒不能同時執行即不能利用多核優勢
很多不懂python的程式設計師會噴python是垃圾 速度太慢 有多核都不能用
反懟:雖然用一個程序下的多個執行緒不能利用多核優勢 但是還可以開設多程序!!!
再次強調:python的多執行緒就是垃圾!!!
反懟:要結合實際情況
如果多個任務都是IO密集型的 那麼多執行緒更有優勢(消耗的資源更少)
多道技術:切換+儲存狀態
如果多個任務都是計算密集型 那麼多執行緒確實沒有優勢 但是可以用多程序
CPU越多越好
以後用python就可以多程序下面開設多執行緒從而達到效率最大化
"""
1.所有的解釋型語言都無法做到同一個程序下多個執行緒利用多核優勢
2.GIL在實際程式設計中其實不用考慮