並發編程（四）

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.)

文檔剖析：

在Cpython中GIL全局解釋器鎖其實也是一把互斥鎖，主要用於阻止同一個進程下的多個執行緒同時被運行(通俗理解：python的多執行緒無法使用多核優勢)；

GIL肯定存在於CPython解釋器中主要原因就在於Cpython解釋器的記憶體管理不是執行緒安全的；

記憶體管理：垃圾回收機制

引用計數

標記清除

分代回收

Cpython解釋器自帶的GIL解釋器鎖，執行緒要想執行程式碼去搶鎖，搶python解釋器，之後才回收，那麼這樣就能保證了阻止同一個進程下的多個執行緒同時被運行，不容易造成數據錯亂；比如，搶票，如果你提交了訂單，那麼別人還能操作到你這張票的訂單嗎？不會了吧；這樣就進而使數據不容易錯亂；

GIL是Cpython解釋器的特點，其實就是一把互斥鎖，犧牲了效率保證了數據的安全（就適用場景而言）；
python同一個進程內的多個執行緒無法利用多核優勢(不能並行但是可以並發（切換加保存狀態）)；
執行緒是執行單位，但是不能直接運行，同一個進程內的多個執行緒要想運行必須先搶GIL鎖，然後拿到python解釋器才能被cpu執行；
GIL的存在其實也解決了垃圾回收機制導致數據錯亂的因素，比如你剛執行了name='hz'，還沒有來得急綁定關係，垃圾回收機制就可能給你回收了，因為垃圾回收也是執行緒，想要執行也得拿解釋器來執行，但是不是和你的程式碼串列；
所有的解釋型語言幾乎都無法實現同一個進程下的多個執行緒同時被運行，但是可以並發；
多個進程下的執行緒才能實現並行；

驗證GIL存在

驗證之前需要明白什麼是多道技術（切換+保存狀態）

多道技術什麼時候切換：程式執行時間長、程式有IO操作(√，示例利用IO)

from threading import Thread
import time
m = 100
def test():
    global m
    tmp = m
    # time.sleep(1)  
    # IO操作，造成數據錯亂，sleep(1)運行釋放了GIL鎖，100個執行緒同樣的操作反覆執行，導致結果為99，如果沒有IO操作結果為0
    tmp -= 1
    m = tmp
for i in range(100):
    t = Thread(target=test)
    t.start()
time.sleep(3)
print(m) # 0

'''
同一個進程下的多個執行緒雖然有GIL的存在不會出現並行的效果,但是如果執行緒內有IO操作還是會造成數據的錯亂，這個時候需要我們額外的添加互斥鎖（就不止GIL一把鎖了）
'''

補：搶鎖釋放鎖簡寫方式

a = Lock()
# 方式一：
a.acquire()
'''程式碼體'''
a.release()
# 方式二：
with a:
    '''程式碼體'''
# 適用with上下文管理器，會自動搶鎖釋放鎖

GIL與普通互斥鎖的區別

from threading import Thread,Lock
import time


mutex = Lock()
m = 100
def test():
    global m
    with mutex:
        tmp = m
        time.sleep(0.1)
        # IO操作，造成數據錯亂，sleep(1)運行釋放了GIL鎖，100個執行緒同樣的操作反覆執行，導致結果為99，如果沒有IO操作結果為0
        tmp -= 1
        m = tmp
if __name__ == '__main__':
    t_list = []
    for i in range(100):
        t = Thread(target=test)
        t.start()
        t_list.append(t)
    for t in t_list:
        t.join()
print(m) # 0

死鎖現象（了解）

存在多把鎖的情況，會出現死鎖現象；

from threading import Thread, Lock
import time

A = Lock()
B = Lock()
class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()

    def func1(self):
        A.acquire()  # 搶鎖
        print('%s 搶到了A鎖' % self.name)  # 相當於獲取執行緒名稱
        # current_thread().name  獲取執行緒名稱
        B.acquire()
        print('%s 搶到了B鎖' % self.name)
        time.sleep(1)
        B.release()  # 釋放鎖
        print('%s 釋放了B鎖' % self.name)
        A.release()
        print('%s 釋放了A鎖' % self.name)

    def func2(self):
        B.acquire()
        print('%s 搶到了B鎖' % self.name)
        A.acquire()
        print('%s 搶到了A鎖' % self.name)
        A.release()
        print('%s 釋放了A鎖' % self.name)
        B.release()
        print('%s 釋放了B鎖' % self.name)

for i in range(10):
    obj = MyThread()
    obj.start()

過程

1、執行緒1搶A鎖，B鎖，其他執行緒等待；

2、執行緒1釋放了B鎖，其他執行緒等待，因為A鎖沒有釋放；執行緒1釋放了A鎖，其他執行緒才能去func1中搶鎖；

3、執行緒1去func2中搶B鎖，A鎖，其他執行緒搶func1中的A鎖和B鎖，現在鎖還在執行緒1手中，那麼就可能導致卡死現象；

4、通俗理解，這樣就導致了，你要的在我手上，我要的在你手上，比如A和B，A在B家被鎖了，B在A家被鎖了，A和B拿的自己家的鑰匙，但是他們在對方的家中，那麼就死鎖了；

遞歸鎖（了解）

遞歸鎖特點：可以被連續的acquire和release，但是只能被第一個搶到這把鎖執行上述操作，它的內部有一個計數器，每acquire一次計數加一，每release一次計數減一，只要計數不為0，其他人都無法搶鎖；

模組：RLock

Factory function that returns a new reentrant lock.

    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it again
    without blocking; the thread must release it once for each time it has
    acquired it.

from threading import Thread,Lock,RLock
import time


'''
mutexA = Lock()
mutexB = Lock()
相當於開啟了兩把鎖;
'''

# 兩把鎖指向同一個記憶體空間地址，開啟一把鎖
A = B = RLock()


class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()

    def func1(self):
        A.acquire()  # 搶鎖
        print('%s 搶到了A鎖' % self.name)  # 相當於獲取執行緒名稱
        # current_thread().name  獲取執行緒名稱
        B.acquire()
        print('%s 搶到了B鎖' % self.name)
        time.sleep(1)
        B.release()  # 釋放鎖
        print('%s 釋放了B鎖' % self.name)
        A.release()
        print('%s 釋放了A鎖' % self.name)

    def func2(self):
        B.acquire()
        print('%s 搶到了B鎖' % self.name)
        A.acquire()
        print('%s 搶到了A鎖' % self.name)
        A.release()
        print('%s 釋放了A鎖' % self.name)
        B.release()
        print('%s 釋放了B鎖' % self.name)

if __name__ == '__main__':
    for i in range(10):
        obj = MyThread()
        obj.start()

#  遞歸鎖的適用不會死鎖，搶一次鎖計數加1，釋放計數減1，其實就是一把鎖沒有導致混亂

訊號量（了解）

訊號量在不同的階段可能對應不同的技術點；

在並發編程中訊號量指的也是鎖；

通俗理解：

互斥鎖是一個廁所，那麼訊號量就是多個廁所

from threading import  Thread,Semaphore
import time
import random

sm = Semaphore(5) # 括弧內寫幾就代表幾個坑位

def task(name):
    sm.acquire() # 搶鎖
    print(f'{name} is running!')
    # time.sleep(3)
    time.sleep(random.randint(1,5))
    sm.release() # 釋放鎖

if __name__ == '__main__':
    for i in range(20):
        t = Thread(target=task,args=(f'拉屎{i}號',))
        t.start()

# 訊號量可以理解為拉屎的坑位，三個人搶鎖（進入廁所），拉完了就是釋放鎖了；

python多執行緒是否沒用（重點）

如果面試官問你python多執行緒是不是沒用啊？你是不是得分情況回答他，不能直接說有用啊，存在即合理···😂

視情況而定來判斷是否需要多執行緒，看程式的類型；

程式的類型
- IO密集型（常用）：需要用戶給定指令或者回饋來執行，比如部落格頁面，你動網頁它就動，不需要實時更進更新等；
- 計算密集型：需要長時間的計算，比如自動駕駛，是不是需要長時間計算路況，或者匹配路線等；
單核多執行緒牛逼，多核情況下，計算密集型開多進程；IO密集型開多執行緒；
可以開多進程結合多執行緒，哈哈哈要啥有啥😁

驗證案例

IO密集型

比如有四個任務，每個任務耗時10s；

開設多進程沒有太大的優勢，共10s+
- 因為遇到IO操作cpu就不再服務，就需要切換，並且開設進程還需要申請記憶體空間和拷貝程式碼；
開設多執行緒有優勢，10s+
- 不需要消耗額外的資源，只需要一個CPU（單核情況下IO密集型開多執行緒是完全有優勢的，以耗時任務最長的為準！）

from multiprocessing import Process
from threading import Thread
import threading
import os,time
def work():
    time.sleep(2)


if __name__ == '__main__':
    l=[]
    print(os.cpu_count()) #本機為8核
    start=time.time()
    for i in range(400):
        # 開進程
        # p=Process(target=work) # run time is 10.905012845993042大部分時間耗費在創建進程上
        # 開執行緒
        p=Thread(target=work) # run time is 2.061677932739258
        l.append(p)
        p.start()
    for p in l:
        p.join()
    stop=time.time()
    print('run time is %s' %(stop-start))
    
# IO密集型的時候，同樣的任務多執行緒只需要2秒

計算密集型

計算密集型（占著cpu不放），比如有四個任務，每個任務耗時10s；

開設多進程可以利用多核優勢，一共10s+
開設多執行緒無法利用多核優勢，因為每個進程都需要10秒，一共40s+

from multiprocessing import Process
from threading import Thread
import os,time
def work():
    res=0
    for i in range(100000000):
        res*=i
if __name__ == '__main__':
    l=[]
    print(os.cpu_count())  # 本機為8核
    start=time.time()
    for i in range(6):
        p=Process(target=work) # 多進程耗時：run time is 6.857659339904785
        # p=Thread(target=work) # 多執行緒耗時：run time is 24.021770000457764
        l.append(p)
        p.start()
    for p in l:
        p.join()
    stop=time.time()
    print('run time is %s' %(stop-start))
    
# 計算密集型，多進程只需6s，這樣只需看一個cpu計算任務所需的時間，那麼多個cpu同時結束；
# 多執行緒就得排隊來了，執行完一個繼續下一個····

Event事件（了解）

一些進程/執行緒需要等待另外一些進程/執行緒運行完畢之後才能運行；

通俗理解：比如汽車生成車間，肯定得先衝壓零部件，然後焊接這些部件（不能扯一體式衝壓昂😓），然後塗裝製程，最後總裝···

那麼把這些一個個流程比作執行緒或進程，是不是得前面的進行完才能往下走？

案例：

from threading import Thread,Event
import time

event = Event() # 造紅綠燈

def light():
    print('紅燈停')
    time.sleep(3)
    print('綠燈行')
    # 行人走
    event.set()

def people(name):
    print(f'{name} 等紅綠燈')
    event.wait()
    print(f'{name} 可以走了')

if __name__ == '__main__':
    t = Thread(target=light)
    t.start()

    for i in range(20):
        t = Thread(target=people,args=(f'{i}',))
        t.start()

執行緒q（了解）

同一個進程下的多個執行緒數據是共享的，為什麼同一個進程下還會去使用隊列呢？因為隊列是管道和鎖構成的，使用隊列也是為了保證數據得到安全（適用場景自定義）

import queue

# 先進先出的隊列
# q = queue.Queue(3)
# q.put(1)
# q.get()
# q.get_nowait()
# q.full()
# q.empty()

# 後進先出隊列，其實和堆棧大差不差
# q = queue.LifoQueue(3)
# q.put(1)
# q.put(2)
# q.put(3)
# print(q.get()) # 3

# 優先順序隊列：可以給放入隊列中的數據設置進出的優先順序
q = queue.PriorityQueue(4)
q.put((10,'111'))
q.put((100,'222'))
q.put((0,'333'))
q.put((-5,'444'))
print(q.get()) # (-5, '444')

# 數字越小，優先順序越高，put((優先順序，數據))

進程池與執行緒池（掌握）

TCP服務端並發

# 麵條版主體
import socket

server = socket.socket()
server.bind(('127.0.0.1',8080))
server.listen(5)

while True:
    sock,addr = server.accept()
    # 通訊循環
    while True:
        try:
            data = sock.recv(1024)
            if len(data) == 0:break
            sock.send(data.upper()) # 發送大寫
        except ConnectionResetError as e:
            print(e)
            break
    sock.close()

封裝版

from threading import Thread
import socket




# 通訊函數
def communication(sock):
    # 通訊循環
    while True:
        try:
            data = sock.recv(1024)
            if len(data) == 0: break
            sock.send(data.upper())  # 發送大寫
        except ConnectionResetError as e:
            print(e)
            break
    sock.close()

# 連接客戶端函數
def server(ip,port):
    server = socket.socket()
    server.bind((ip,port))
    server.listen(5)
    while True:
        sock,addr = server.accept()

        # 開設多進程/多執行緒
        t = Thread(target=communication,args=(sock,))
        t.start()

if __name__ == '__main__':
    s = Thread(target=server,args=('127.0.0.1',8080))
    s.start()

思考:能否無限制的開設進程或者執行緒???

肯定是不能無限制開設的，如果單從技術層面上來說無限開設肯定是可以的並且是最高效，但是從硬體層面上來說是無法實現的(硬體的發展永遠趕不上軟體的發展速度)

這時候就出現了池，我們在合理適用電腦的時候，保證硬體正常工作的前提，去開設多進程和多執行緒，是最合理的，如果硬體崩潰了，軟體也沒用了；

池：池是用來保證電腦安全情況下，最大限度的利用電腦，它的出現降低了程式的運行效率但是保證了電腦硬體的安全，從而相對高效運行；（對比半連接池只限制了等待的數量；）
進程池：提前開設了固定個數的進程之後反覆調用這些進程完成工作(後續不再開設新的)
執行緒池：提前開設了固定個數的執行緒之後反覆調用這些執行緒完成工作(後續不再開設新的)

進程池與執行緒池的基本使用

模組：from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

執行緒池

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time

# 執行緒池：固定開設5個執行緒，5個執行緒不會重複出現重複創建和銷毀(節省資源);
pool = ThreadPoolExecutor(5) # 括弧內可以傳數字，不傳默認開設當前電腦cpu個數五倍的執行緒
"""Initializes a new ThreadPoolExecutor instance.

        Args:
            max_workers: The maximum number of threads that can be used to
                execute the given calls.
            thread_name_prefix: An optional name prefix to give our threads.
        """


def task(n):
    print(n)
    time.sleep(2)
    return f'返回結果：{n*2}'

# pool.submit(task,1) # 朝池子中提交任務(非同步)
'''
提交方式：同步、非同步
'''
# print('main')


# 朝池子提交20個任務，每次只能接5個
t_list = []
for i in range(20):
    res = pool.submit(task,i)
    # print(res.result())
    # 返回結果，非同步變串列，同步提交
    t_list.append(res)
pool.shutdown()  # 關閉執行緒池，等待執行緒池中所有任務運行完畢
# 解決了等待卡頓
for t in t_list: # 非同步提交結果，先起任務再返回結果
    print('>>>>',t.result())

進程池

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time
import os

# 進程池：固定開設幾個進程，進程不會重複出現重複創建和銷毀(節省資源);
pool = ProcessPoolExecutor(5) # 括弧內可以傳數字，不傳默認開設當前電腦cpu個數

def task(n):
    print(n,os.getpid())
    time.sleep(2)
    return f'進程號：{os.getpid()}'

# 非同步提交任務的返回結果，應該通過回調機制來獲取，而不是下面的for循環最後獲取
def call_back(n): # n 返回的對象
    print(n.result())   # n.result 相當於res.result
if __name__ == '__main__':

    # 朝池子提交20個任務，每次只能接5個
    # t_list = []
    for i in range(20):
        res = pool.submit(task,i).add_done_callback(call_back)   # 回調機制
        # print(res.result())
        # 返回結果，非同步變串列，同步提交
        # t_list.append(res)
    # pool.shutdown()
    # # 解決了等待卡頓
    # for t in t_list: # 非同步提交結果，先起任務再返回結果
    #     print('>>>>',t.result())

主要操作：

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
pool = ProcessPoolExecutor(5)
res = pool.submit(task,i).add_done_callback(call_back)   # 回調機制

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
import os

# 創建進程池與執行緒池
# pool = ThreadPoolExecutor(5)  # 可以自定義執行緒數 也可以採用默認策略
pool = ProcessPoolExecutor(5)  # 可以自定義執行緒數 也可以採用默認策略


# 定義一個任務
def task(n):
    print(n, os.getpid())
    time.sleep(2)
    return '>>>:%s' % n ** 2


# 定義一個回調函數:非同步提交完之後有結果自動調用該函數
def call_back(a):
    print('非同步回調函數:%s' % a.result())


# 朝執行緒池中提交任務
# obj_list = []
for i in range(20):
    res = pool.submit(task, i).add_done_callback(call_back)  # 非同步提交
    # obj_list.append(res)
"""
同步:提交完任務之後原地等待任務的返回結果 期間不做任何事
非同步:提交完任務之後不願地等待任務的返回結果 結果由非同步回調機制自動回饋
"""
# 等待執行緒池中所有的任務執行完畢之後 再獲取各自任務的結果
# pool.shutdown()
# for i in obj_list:
#     print(i.result())  # 獲取任務的執行結果  同步

在windows電腦中如果是進程池的使用也需要在__main__下面

協程

名詞解釋

進程：資源單位
執行緒：工作單位
協程：程式設計師自定義的名詞，意思是單執行緒下實現並發（程式設計師自己在程式碼層面上監測我們所有的IO操作，一但遇到IO，我們在程式碼級別完成切換，這樣給cpu的感覺是程式一直在運行，沒有IO操作從而提升效率）
多道技術：切換+保存技術
CPU被剝奪的條件：
- 程式長時間佔用
- 程式進入IO操作
並發去實現切換+保存狀態

欺騙CPU的行為：

單執行緒下我們如果能夠自己檢測IO操作並且自己實現程式碼層面的切換
那麼對於CPU而言我們這個程式就沒有IO操作，CPU會儘可能的被佔用

注意切換不一定能提升效率，如果是IO密集型就會提升效率，計算密集型切換就會降低效率；

gevent模組

能夠自主監測IO行為並切換

from gevent import monkey;monkey.patch_all()
# 固定程式碼格式加上之後才能檢測所有的IO行為
from gevent import spawn
import time


def play(name):
    print('%s play 1' % name)
    time.sleep(5)
    print('%s play 2' % name)
'''
兩個方法有IO操作，程式碼一直在反覆「跳」
'''

def eat(name):
    print('%s eat 1' % name)
    time.sleep(3)
    print('%s eat 2' % name)


start = time.time()
# play('Hammer')  # 正常的同步調用
# eat('Hammer')  # 正常的同步調用 8s+
g1 = spawn(play, 'Hammer')  # 非同步提交
g2 = spawn(eat, 'Hammer')  # 非同步提交
g1.join()
g2.join()  # 等待被監測的任務運行完畢
print('主', time.time() - start)  # 單執行緒下實現並發，提升效率5s+

協程實現TCP服務端並發的效果

# 並發效果:一個服務端可以同時服務多個客戶端
import socket
from gevent import monkey;monkey.patch_all()
from gevent import spawn
def talk(sock):
    while True:
        try:
            data = sock.recv(1024)
            if len(data) == 0:break
            print(data)
            sock.send(data+b'hi')
        except ConnectionResetError as e:
            print(e)
            sock.close()
            break
def servers():
    server = socket.socket()
    server.bind(('127.0.0.1',8080))
    server.listen()
    while True:
        sock, addr = server.accept()
        spawn(talk,sock)
g1 = spawn(servers)
g1.join()

# 客戶端開設幾百個執行緒發消息即可
最牛的情況:多進程下開設多執行緒，多執行緒下開設協程
我們以後可能自己動手寫的不多，一般都是使用別人封裝好的模組或框架

IO模型

IO模型研究的主要是網路IO(linux系統)，理論為主，程式碼實現大部分為偽程式碼；

基本關鍵字

同步（synchronous）
非同步（asynchronous）
阻塞（blocking）
非阻塞（non-blocking）

研究方向

Stevens在文章中一共比較了五種IO Model
blocking IO 阻塞IO
nonblocking IO 非阻塞IO
IO multiplexing IO多路復用
signal driven IO 訊號驅動IO
asynchronous IO 非同步IO

由signal driven IO（訊號驅動IO）在實際中並不常用，所以主要介紹其餘四種IO Model

四種IO模型介紹

阻塞IO

最為常見的一種IO模型，有兩個等待的階段(wait for data、copy data)

電腦1和電腦2數據傳輸，需要經過拷貝到記憶體到OSI，然後才到電腦2的OSI七層到記憶體；

非阻塞IO

系統調用階段變為了非阻塞(輪詢) 有一個等待的階段(copy data)，輪詢的階段是比較消耗資源的；

通俗的理解：會一直詢問kernel有沒有數據~，並不會阻塞操作，直到copy才會阻塞；

多路復用IO

利用select或者epoll來監管多個程式一旦某個程式需要的數據存在於記憶體中了那麼立刻通知該程式去取即可；

當用戶進程調用了select，那麼整個進程會被block，而同時，kernel會「監視」所有select負責的socket，當任何一個socket中的數據準備好了，select就會返回。這個時候用戶進程再調用read操作，將數據從kernel拷貝到用戶進程；通俗理解為：多個人排隊取餐，監控select，如果參號了，kernel說好了，然後用戶去取餐return；

這個圖和blocking IO的圖其實並沒有太大的不同，事實上還更差一些。因為這裡需要使用兩個系統調用(select和recvfrom)，而blocking IO只調用了一個系統調用(recvfrom)。但是，用select的優勢在於它可以同時處理多個connection；

非同步IO

只需要發起一次系統調用之後無需頻繁發送有結果並準備好之後會通過非同步回調機制回饋給調用者；

Tags: 網路編程並發編編程（四）

並發編程（四）

GIL全局解釋器鎖（重點）

驗證GIL存在

GIL與普通互斥鎖的區別

死鎖現象（了解）

過程

遞歸鎖（了解）

訊號量（了解）

python多執行緒是否沒用（重點）

驗證案例

IO密集型

計算密集型

Event事件（了解）

執行緒q（了解）

進程池與執行緒池（掌握）

進程池與執行緒池的基本使用

執行緒池

進程池

協程

名詞解釋

gevent模組

協程實現TCP服務端並發的效果

IO模型

基本關鍵字

研究方向

四種IO模型介紹

阻塞IO

非阻塞IO

多路復用IO

非同步IO

VirMach 便宜 VPS

QNews

並發編程（四）

GIL全局解釋器鎖（重點）

驗證GIL存在

GIL與普通互斥鎖的區別

死鎖現象（了解）

過程

遞歸鎖（了解）

訊號量（了解）

python多執行緒是否沒用（重點）

驗證案例

IO密集型

計算密集型

Event事件（了解）

執行緒q（了解）

進程池與執行緒池（掌握）

進程池與執行緒池的基本使用

執行緒池

進程池

協程

名詞解釋

gevent模組

協程實現TCP服務端並發的效果

IO模型

基本關鍵字

研究方向

四種IO模型介紹

阻塞IO

非阻塞IO

多路復用IO

非同步IO

分享此文：

Related Posts

☕【Java深層系列】「並發編程系列」讓我們一起探索一下CyclicBarrier的技術原理和源碼分析

資料庫原理（一） 資料庫基本概念

鎚子科技的困難終於全沒了！羅永浩即將重返科技行業

改變行業的電競筆電升級了！雙顯三模、140W滿血

VirMach 便宜 VPS

QNews

熱門文章

熱門搜尋

資料庫原理（一）資料庫基本概念