併發程式設計 03

阿新 • • 發佈：2021-09-03

併發程式設計

昨日內容回顧

程序物件屬性及其他方法

from multiprocessing import Process，current_process
import os
current_process().pid   #檢視當前程序號
os.getpid() #檢視當前程序號
os.getppid()#檢視當前程序的父程序號
windows中終端命令
tasklist
tasklist|findstr pid

僵死程序與孤兒程序

僵死程序
孤兒程序

守護程序

被守護程序結束後守護程序立刻也跟著結束
如何開啟：在start之前寫p.daemon=True
                     p.start()

互斥鎖

"""
多個人在操作同一份資料的時候，可能會出現資料錯亂的問題
針對上述問題我們正常都是加鎖處理
作用：將併發變成序列，犧牲了程式執行的效率但是保證了資料的安全
注意：只在操作資料的部分加鎖即可
鎖儘量不要自己去處理，很容易造成死鎖現象
"""
擴充套件：行鎖，表鎖
#搶鎖
q.acquire()
#釋放鎖
q.release()
#模擬搶鎖

佇列Queue

佇列：先進先出
堆疊：先進後出
from multiprocessing import Queue
q=Queue()
q.put()
q.get()
q.full()
q.empty()
q.get_nowait()#取資料的時候如果沒有資料直接報錯
q.get(timeout=5)

程序間通訊

程序間無法直接進行資料互動，但是可以通過佇列或者管道實現資料互動
本地測試的時候可能會用到Queue，實際生產用的都是別人封裝好的功能非常強大的工具

生產者與消費者模型

生產者+訊息佇列+消費者
為何要有訊息佇列，是為了解決供需不平衡的問題
#joinableQueue
可以被等待的q
你在往佇列中放資料的時候，內部會有一個計數器自動加1
你在往佇列中放資料的時候，呼叫task_done(),內部計數器自動減1
q.join()  當計數器為0的時候才繼續往下執行

執行緒理論

程序：資源單位
執行緒：執行單位。執行緒是真正幹活的人，幹活的過程中所需的資源由執行緒所在的程序提供
每一個程序肯定都自帶一個執行緒
同一個程序內可以建立多個執行緒
開程序：申請記憶體空間，拷貝程式碼，消耗資源較大
開執行緒：同一個程序內建立多個執行緒，無需上述2個步驟，消耗資源相對較小

今日內容

開啟執行緒的2種方式

#開啟執行緒不需要在main下面執行程式碼，直接書寫就可以
#但是我們習慣性的將啟動寫在main下
#針對兩個雙下劃線開頭雙下劃線開頭的方法，讀成雙下init
from multiprocessing import Process
from threading import Thread
def task():
    print("子執行緒")
if __name__ == '__main__':
    t = Process(target=task)
    t.start()
    print("主執行緒")
# from multiprocessing import Process
# from threading import Thread
# def task():
#     print("子執行緒")
# if __name__ == '__main__':
#     t=Thread(target=task)
#     t.start()
#     print("主執行緒")

如何實現tcp服務端併發的效果

import socket
from threading import Thrread
from multiprocess import Process
#服務端要有固定的埠 24小時不間斷提供服務 能夠支援併發
server=socket.socket() #括號內不加引數預設是tcp協議
server.bind(("127.0.0.1",8080))
server.listen(5)
#連結迴圈
def talk(conn):
    while True:
        try:
            data=conn.recv(1024)
            if len(data)==0:break
            print(data.decode("utf-8"))
            conn.send(data.upper())
        except ConnectionResetError as e
            print(e)
            break
     conn,close()
while True:
    conn,addr=server.accept()
    t=Thread(target=talk,args=(conn,))
    t.start()
 #客戶端
import socket
client=socket.socket()
client.connect(("127.0.0.1",8080))
while True:
    client.send(b"hello")
    data=client.recv(1024)
    print(data.decode("utf-8"))

執行緒物件的join方法

t.join  同進程

同一個程序下的多個執行緒資料共享

from threading import Thread
import time
money=100
def task():
    global money
    money=666
if __name__ == '__main__':
    t=Thread(target=task)
    t.start()
    print(money)

執行緒物件及其他方法

rom threading import Thread,active_count,current_thread
import time,os

def task():
    #print("hello",os.getpid())
    print("hello",current_thread().name)


if __name__ == '__main__':
    t=Thread(target=task)
    t.start()
    print("主",active_count())
    #print("主",os.getpid())
    print("主",current_thread().name)

守護執行緒

主執行緒執行結束之後不會立即結束，會等待其他非守護執行緒結束才結束
因為主執行緒的結束意味著所在程序的結束

t.setDaemon()

執行緒互斥鎖

from threading import Thread,Lock
import time
money=100
lock=Lock()
def task():
    global money
    lock.acquire()
    tmp=money
    time.sleep(0.1)
    money=tmp-1
    lock.release()
if __name__ == '__main__':

    t_list=[]
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t_list.append(t)
    for t in t_list:
        t.join()
    print(money)

GIL全域性直譯器鎖

python直譯器有多個版本，cpython,pypypython,但普遍使用的是cpython
在cpython直譯器中GIL是一把互斥鎖，用來阻止同一個程序下的多個執行緒的同時執行
同一程序下的多個執行緒無法利用多核優勢
因為cpython中的記憶體管理不是執行緒安全的
記憶體管理（垃圾回收機制）
1.引用計數
2.標記清楚
3.分代回收
重點：1.GIL不是python的特點，而是cpython直譯器的特點
      2.GIL是保證直譯器級別的資料安全
      3.同一程序下的多個執行緒無法同時進行 ，即無法利用多核優勢
      4.針對不同的資料還是需要加不同的鎖處理
    5.解釋型語言的通病同意程序下多個執行緒無法利用多核優勢

GIL與普通鎖的區別

同一個程序下的多執行緒無法利用多核優勢，是不是就沒有用了

多執行緒分單核多核

計算密集型：單核：多程序：額外的消耗資源

多執行緒：

多核：多程序：總耗時10+

多執行緒：總耗時40+

I/O密集型：多核：多程序：相對資源浪費

多執行緒：更加節省資源

### 死鎖與遞迴鎖

死鎖案列
import time

from threading import Thread,Lock,RLock
lock1 = Lock()
lock2 = Lock()
# lock1 = RLock()  # 遞迴鎖， 可重入鎖
# lock2 = RLock()

def eat1( name):
    lock1.acquire()
    print("%s搶到了麵條" % name)
    time.sleep(1)
    lock2.acquire()
    print("%s搶到了筷子" % name)
    time.sleep(1)
    lock2.release()
    print("%s放下了筷子" % name)
    time.sleep(1)
    print("%s放下了麵條" % name)
    lock1.release()


def eat2(name):
    lock2.acquire()
    print("%s搶到了筷子" % name)
    time.sleep(1)
    lock1.acquire()
    print("%s搶到了麵條" % name)
    time.sleep(1)
    lock1.release()
    print("%s放下了麵條" % name)
    time.sleep(1)
    print("%s放下了筷子" % name)
    lock2.release()

if __name__ == '__main__':

    for name in ['egon', 'jason', 'ly']:
        t = Thread(target=eat1, args=( name,))
        t.start()

    for name in ['qq', 'tom', 'kevin']:
        t1 = Thread(target=eat2, args=(name,))
        t1.start()

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

訊號量

訊號量在不同的階段可能對應不同的技術點
在併發程式設計中訊號量指的是鎖
如果我們將互斥鎖比喻程一個廁所的話，那麼訊號量就相當於多個廁所
sm=semaphore(5)#括號內寫數字，寫幾就表示開設幾個坑位
from threading import Thread, Semaphore
from multiprocessing import Process, Lock
# Semaphore:訊號量可以理解為多把鎖，同時允許多個執行緒來更改資料

import time, random

sm = Semaphore(2)


def task(i):
    sm.acquire()
    print("執行緒：%s,進來了" % i)
    time.sleep(random.randint(1, 3))
    print("執行緒：%s,出去了" % i)
    sm.release()


if __name__ == '__main__':
    for i in range(6):
        t = Thread(target=task, args=(i,))
        t.start()

Event事件

一些執行緒或者程序需要等待另外一些程序/執行緒執行完畢之後才能執行，類似於發射訊號一樣

from threading import Thread,Event
import time
event=Event()   #造了一個紅綠燈
def light():
    print("紅燈亮著的")
    time.sleep(3)
    print("綠燈亮了")
    event.set()
def car(name):
    print("%s 車正在等紅燈" %name)
    event.wait()
    print("%s 車加油門開走了" %name)
if __name__=="__main__":
    t=Thread(target=light)
    t.start()
    for i in range(10):
        t=Thread(target=car,args=("%s" %i))
        t.start()