python高級之多線程

本節內容

1. 線程與進程定義及區別
2. python全局解釋器鎖
3. 線程的定義及使用
4. 互斥鎖
5. 線程死鎖和遞歸鎖
6. 條件變量同步(Condition)
7. 同步條件(Event)
8. 信號量
9. 隊列Queue
10. Python中的上下文管理器(contextlib模塊)
11. 自定義線程池

1.線程與進程定義及區別

線程的定義：

線程是操作系統能夠進行運算調度的最小單位。它被包含在進程之中，是進程中的實際運作單位。一條線程指的是進程中一個單一順序的控制流，一個進程中可以並發多個線程，每條線程並行執行不同的任務。

進程的定義：

進程（Process）是計算機中的程序關於某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操作系統結構的基礎。在早期面向進程設計的計算機結構中，進程是程序的基本執行實體。程序是指令、數據及其組織形式的描述，進程是程序的實體。

進程和線程的區別

• Threads share the address space of the process that created it; processes have their own address space.
• 線程的地址空間共享，每個進程有自己的地址空間。
• Threads have direct access to the data segment of its process; processes have their own copy of the data segment of the parent process.
• 一個進程中的線程直接接入他的進程的數據段，但是每個進程都有他們自己的從父進程拷貝過來的數據段
• Threads can directly communicate with other threads of its process; processes must use interprocess communication to communicate with sibling processes.
• 一個進程內部的線程之間能夠直接通信，進程之間必須使用進程間通信實現通信
• New threads are easily created; new processes require duplication of the parent process.
• 新的線程很容易被創建，新的進程需要從父進程復制
• Threads can exercise considerable control over threads of the same process; processes can only exercise control over child processes.
• 一個進程中的線程間能夠有相當大的控制力度，進程僅僅只能控制他的子進程
• Changes to the main thread (cancellation, priority change, etc.) may affect the behavior of the other threads of the process; changes to the parent process does not affect child processes.
• 改變主線程（刪除，優先級改變等）可能影響這個進程中的其他線程；修改父進程不會影響子進程

2.python全局解釋器鎖

全局解釋器鎖又叫做GIL

python目前有很多解釋器，目前使用最廣泛的是CPython，還有PYPY和JPython等解釋器，但是使用最廣泛的還是CPython解釋器，而對於全局解釋器鎖來說，就是在CPython上面才有的，它的原理是在解釋器層面加上一把大鎖，保證同一時刻只能有一個python線程在解釋器中執行。

對於計算密集型的python多線程來說，無法利用到多線程帶來的效果， 在2.7時計算密集型的python多線程執行效率比順序執行的效率還低的多，在python3.5中對這種情況進行了優化，基本能實現這種多線程執行時間和順序執行時間差不多的效果。

對於I/O密集型的python多線程來說，GIL的影響不是很大，因為I/O密集型的python多線程進程，每個線程在等待I/O的時候，將會釋放GIL資源，供別的線程來搶占。所以對於I/O密集型的python多線程進程來說，還是能比順序執行的效率要高的。

python的GIL這個東西。。。比較惡心，但是由於CPython解釋器中的很多東西都是依賴這個東西開發的，如果改的話，將是一件浩大的工程。。。所以到現在還是存在這個問題，這也是python最為別人所詬病的地方。。。

3.線程的定義及使用

線程的兩種調用方式

線程的調用有兩種方式，分為直接調用和繼承式調用，示例代碼如下：

#直接調用
import threading
import time

def sayhi(num): #定義每個線程要運行的函數

    print("running on number:%s" %num)

    time.sleep(3)

if __name__ == '__main__':

    t1 = threading.Thread(target=sayhi,args=(1,)) #生成一個線程實例
    t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一個線程實例

    t1.start() #啟動線程
    t2.start() #啟動另一個線程

    print(t1.getName()) #獲取線程名
    print(t2.getName())

#繼承式調用
import threading
import time


class MyThread(threading.Thread):
    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):#定義每個線程要運行的函數

        print("running on number:%s" %self.num)

        time.sleep(3)

if __name__ == '__main__':

    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

 

可以看到直接調用是導入threading模塊並定義一個函數，之后實例化threading.Thread類的時候，將剛定義的函數名通過target參數傳遞進去，然后調用實例的start()方法啟動一個線程。

而繼承式調用是創建一個類繼承自threading.Thread類，並在構造方法中調用父類的構造方法，之后重寫run方法，run方法中就是每個線程起來之后執行的內容，就類似於前面通過target參數傳遞進去的函數。之后以這個繼承的類來創建對象，並執行對象的start()方法啟動一個線程。

從這里可以看出，其實。。。直接調用通過使用target參數把函數帶進類里面之后應該是用這個函數替代了run方法。

join和setDaemon

join()方法在該線程對象啟動了之后調用線程的join()方法之后，那么主線程將會阻塞在當前位置直到子線程執行完成才繼續往下走，如果所有子線程對象都調用了join()方法，那么主線程將會在等待所有子線程都執行完之后再往下執行。

setDaemon(True)方法在子線程對象調用start()方法（啟動該線程）之前就調用的話，將會將該子線程設置成守護模式啟動，這是什么意思呢？當子線程還在運行的時候，父線程已經執行完了，如果這個子線程設置是以守護模式啟動的，那么隨着主線程執行完成退出時，子線程立馬也退出，如果沒有設置守護啟動子線程（也就是正常情況下）的話，主線程執行完成之后，進程會等待所有子線程執行完成之后才退出。

示例代碼如下：

import threading
from time import ctime,sleep
import time

def music(func):
    for i in range(2):
        print ("Begin listening to %s. %s" %(func,ctime()))
        sleep(4)
        print("end listening %s"%ctime())

def move(func):
    for i in range(2):
        print ("Begin watching at the %s! %s" %(func,ctime()))
        sleep(5)
        print('end watching %s'%ctime())

threads = []
t1 = threading.Thread(target=music,args=('七里香',))
threads.append(t1)
t2 = threading.Thread(target=move,args=('阿甘正傳',))
threads.append(t2)

if __name__ == '__main__':

    for t in threads:
        # t.setDaemon(True)
        t.start()
        # t.join()
    # t1.join()
    t2.join()########考慮這三種join位置下的結果？
    print ("all over %s" %ctime())

 

4.互斥鎖

互斥鎖的產生是因為前面提到過多線程之間是共享同一塊內存地址的，也就是說多個不同的線程能夠訪問同一個變量中的數據，那么，當多個線程要修改這個變量，會產生什么情況呢？當多個線程修改同一個數據的時候，如果操作的時間夠短的話，能得到我們想要的結果，但是，如果修改數據不是原子性的（這中間的時間太長）的話。。。很有可能造成數據的錯誤覆蓋，從而得到我們不想要的結果。例子如下：

import time
import threading

def addNum():
    global num #在每個線程中都獲取這個全局變量
    # num-=1  # 如果是這種情況，那么操作時間足夠短，類似於原子操作了，所以，能夠得到我們想要的結果

    temp=num
    print('--get num:',num )  # 因為print會調用終端輸出，終端是一個設備，相當於要等待終端I/O就緒之后才能輸出打印內容，在等待終端I/O的過程中，該線程已經掛起。。。這時其他線程獲取到的是沒被改變之前的num值，之后該線程I/O就緒之后切換回來，對num-1了，其他線程在I/O就緒之后也在沒被改變之前的num基礎上減一，這樣。。。就得到了我們不想看到的結果。。。
    #time.sleep(0.1)  # sleep也能達到相同的效果，執行到sleep時，該線程直接進入休眠狀態，釋放了GIL直到sleep時間過去。
    num =temp-1 #對此公共變量進行-1操作


num = 100  #設定一個共享變量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)

for t in thread_list: #等待所有線程執行完畢
    t.join()

print('final num:', num )
這時候，就需要互斥鎖出場了，前面出現的num可以被稱作臨界資源（會被多個線程同時訪問），為了讓臨界資源能夠實現按照我們控制訪問，需要使用互斥鎖來鎖住臨界資源，當一個線程需要訪問臨界資源時先檢查這個資源有沒有被鎖住，如果沒有被鎖住，那么訪問這個資源並同時給這個資源加上鎖，這樣別的線程就無法訪問該臨界資源了，直到這個線程訪問完了這個臨界資源之后，釋放這把鎖，其他線程才能夠搶占該臨界資源。這個，就是互斥鎖的概念。 示例代碼：

import time
import threading

def addNum():
    global num #在每個線程中都獲取這個全局變量
    # num-=1
    lock.acquire()  # 檢查互斥鎖，如果沒鎖住，則鎖住並往下執行，如果檢查到鎖住了，則掛起等待鎖被釋放時再搶占。
    temp=num
    print('--get num:',num )
    #time.sleep(0.1)
    num =temp-1 #對此公共變量進行-1操作
    lock.release()  # 釋放該鎖

num = 100  #設定一個共享變量
thread_list = []
lock=threading.Lock()  # 定義互斥鎖

for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)

for t in thread_list: #等待所有線程執行完畢
    t.join()

print('final num:', num )

 

互斥鎖與GIL的關系？

Python的線程在GIL的控制之下，線程之間，對整個python解釋器，對python提供的C API的訪問都是互斥的，這可以看作是Python內核級的互斥機制。但是這種互斥是我們不能控制的，我們還需要另外一種可控的互斥機制———用戶級互斥。內核級通過互斥保護了內核的共享資源，同樣，用戶級互斥保護了用戶程序中的共享資源。

GIL 的作用是：對於一個解釋器，只能有一個thread在執行bytecode。所以每時每刻只有一條bytecode在被執行一個thread。GIL保證了bytecode 這層面上是thread safe的。 但是如果你有個操作比如 x += 1，這個操作需要多個bytecodes操作，在執行這個操作的多條bytecodes期間的時候可能中途就換thread了，這樣就出現了data races的情況了。

5.線程死鎖和遞歸鎖

如果公共的臨界資源比較多，並且線程間都使用互斥鎖去訪問臨界資源，那么將有可能出現一個情況：

• 線程1拿到了資源A，接着需要資源B才能繼續執行下去
• 線程2拿到了資源B，接着需要資源A才能繼續執行下去

這樣，線程1和線程2互不相讓。。。結果就都卡死在這了，這就是線程死鎖的由來。。。

示例代碼如下：

import threading,time

class myThread(threading.Thread):
    def doA(self):
        lockA.acquire()  # 鎖住A資源
        print(self.name,"gotlockA",time.ctime())
        time.sleep(3)
        lockB.acquire()  # 鎖住B資源
        print(self.name,"gotlockB",time.ctime())
        lockB.release()  # 解鎖B資源
        lockA.release()  # 解鎖A資源

    def doB(self):
        lockB.acquire()
        print(self.name,"gotlockB",time.ctime())
        time.sleep(2)
        lockA.acquire()
        print(self.name,"gotlockA",time.ctime())
        lockA.release()
        lockB.release()
    def run(self):
        self.doA()
        self.doB()
if __name__=="__main__":

    lockA=threading.Lock()
    lockB=threading.Lock()
    threads=[]
    for i in range(5):
        threads.append(myThread())
    for t in threads:
        t.start()
    for t in threads:
        t.join()  # 等待線程結束

 

那么，怎么解決這個問題呢？python中提供了一個方法（不止python中，基本上所有的語言中都支持這個方法）那就是遞歸鎖。遞歸鎖的創建是使用threading.RLock()，它里面其實維護了兩個東西，一個是Lock，另一個是counter，counter記錄了加鎖的次數，每加一把鎖，counter就會+1，釋放一次鎖counter就會減一，直到所有加的鎖都被釋放掉了之后其他線程才能夠訪問這把鎖獲取資源。當然這個限制是對於線程之間的，同一個線程中，只要這個線程搶到了這把鎖，那么這個線程就可以對這把鎖加多個鎖，而不會阻塞自己的執行。這就是遞歸鎖的原理。

示例代碼：

import time
import threading
class Account:
    def __init__(self, _id, balance):
        self.id = _id
        self.balance = balance
        self.lock = threading.RLock()

    def withdraw(self, amount):
        with self.lock:  # 會將包裹的這塊代碼用鎖保護起來，直到這塊代碼執行完成之后，這把鎖就會被釋放掉
            self.balance -= amount

    def deposit(self, amount):
        with self.lock:
            self.balance += amount


    def drawcash(self, amount):#lock.acquire中嵌套lock.acquire的場景

        with self.lock:
            interest=0.05
            count=amount+amount*interest

            self.withdraw(count)


def transfer(_from, to, amount):

    #鎖不可以加在這里 因為其他的線程執行的其它方法在不加鎖的情況下數據同樣是不安全的
     _from.withdraw(amount)
     to.deposit(amount)

alex = Account('alex',1000)
yuan = Account('yuan',1000)

t1=threading.Thread(target = transfer, args = (alex,yuan, 100))
t1.start()

t2=threading.Thread(target = transfer, args = (yuan,alex, 200))
t2.start()

t1.join()
t2.join()

print('>>>',alex.balance)
print('>>>',yuan.balance)

 

6.條件變量同步(Condition)

有一類線程需要滿足條件之后才能夠繼續執行，Python提供了threading.Condition 對象用於條件變量線程的支持，它除了能提供RLock()或Lock()的方法外，還提供了 wait()、notify()、notifyAll()方法。

lock_con=threading.Condition([Lock/Rlock])： 鎖是可選選項，不傳人鎖，對象自動創建一個RLock()。

1. wait()：條件不滿足時調用，線程會釋放鎖並進入等待阻塞；
2. notify()：條件創造后調用，通知等待池激活一個線程；
3. notifyAll()：條件創造后調用，通知等待池激活所有線程。

示例代碼：

import threading,time
from random import randint
class Producer(threading.Thread):
    def run(self):
        global L
        while True:
            val=randint(0,100)
            print('生產者',self.name,":Append"+str(val),L)
            if lock_con.acquire():
                L.append(val)
                lock_con.notify()
                lock_con.release()
            time.sleep(3)
class Consumer(threading.Thread):
    def run(self):
        global L
        while True:
                lock_con.acquire()
                if len(L)==0:
                    lock_con.wait()
                print('消費者',self.name,":Delete"+str(L[0]),L)
                del L[0]
                lock_con.release()
                time.sleep(0.25)

if __name__=="__main__":

    L=[]
    lock_con=threading.Condition()
    threads=[]
    for i in range(5):
        threads.append(Producer())
    threads.append(Consumer())
    for t in threads:
        t.start()
    for t in threads:
        t.join()

 

7.同步條件(Event)

條件同步和條件變量同步差不多意思，只是少了鎖功能，因為條件同步設計於不訪問共享資源的條件環境。event=threading.Event()：條件環境對象，初始值 為False；

1. event.isSet()：返回event的狀態值；
2. event.wait()：如果 event.isSet()==False將阻塞線程；
3. event.set()： 設置event的狀態值為True，所有阻塞池的線程激活進入就緒狀態， 等待操作系統調度；
4. event.clear()：恢復event的狀態值為False。

例子1：

import threading,time
class Boss(threading.Thread):
    def run(self):
        print("BOSS：今晚大家都要加班到22:00。")
        event.isSet() or event.set()
        time.sleep(5)
        print("BOSS：<22:00>可以下班了。")
        event.isSet() or event.set()
class Worker(threading.Thread):
    def run(self):
        event.wait()
        print("Worker：哎……命苦啊！")
        time.sleep(0.25)
        event.clear()
        event.wait()
        print("Worker：OhYeah!")
if __name__=="__main__":
    event=threading.Event()
    threads=[]
    for i in range(5):
        threads.append(Worker())
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()

 

例子2：

import threading,time
import random
def light():
    if not event.isSet():
        event.set() #wait就不阻塞 #綠燈狀態
    count = 0
    while True:
        if count < 10:
            print('\033[42;1m--green light on---\033[0m')
        elif count <13:
            print('\033[43;1m--yellow light on---\033[0m')
        elif count <20:
            if event.isSet():
                event.clear()
            print('\033[41;1m--red light on---\033[0m')
        else:
            count = 0
            event.set() #打開綠燈
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(random.randrange(10))
        if  event.isSet(): #綠燈
            print("car [%s] is running.." % n)
        else:
            print("car [%s] is waiting for the red light.." %n)
if __name__ == '__main__':
    event = threading.Event()
    Light = threading.Thread(target=light)
    Light.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

 

8.信號量

信號量用來控制線程並發數的，BoundedSemaphore或Semaphore管理一個內置的計數 器，每當調用acquire()時-1，調用release()時+1。

計數器不能小於0，當計數器為 0時，acquire()將阻塞線程至同步鎖定狀態，直到其他線程調用release()。(類似於停車位的概念)

BoundedSemaphore與Semaphore的唯一區別在於前者將在調用release()時檢查計數 器的值是否超過了計數器的初始值，如果超過了將拋出一個異常。

例子：

import threading,time
class myThread(threading.Thread):
    def run(self):
        if semaphore.acquire():
            print(self.name)
            time.sleep(5)
            semaphore.release()
if __name__=="__main__":
    semaphore=threading.Semaphore(5)
    thrs=[]
    for i in range(100):
        thrs.append(myThread())
    for t in thrs:
        t.start()

 

9.隊列Queue

使用隊列方法：

創建一個“隊列”對象
import Queue
q = Queue.Queue(maxsize = 10)
Queue.Queue類即是一個隊列的同步實現。隊列長度可為無限或者有限。可通過Queue的構造函數的可選參數maxsize來設定隊列長度。如果maxsize小於1就表示隊列長度無限。

將一個值放入隊列中
q.put(10)
調用隊列對象的put()方法在隊尾插入一個項目。put()有兩個參數，第一個item為必需的，為插入項目的值；第二個block為可選參數，默認為
1。如果隊列當前為空且block為1，put()方法就使調用線程暫停,直到空出一個數據單元。如果block為0，put方法將引發Full異常。

將一個值從隊列中取出
q.get()
調用隊列對象的get()方法從隊頭刪除並返回一個項目。可選參數為block，默認為True。如果隊列為空且block為True，get()就使調用線程暫停，直至有項目可用。如果隊列為空且block為False，隊列將引發Empty異常。

Python Queue模塊有三種隊列及構造函數:
1、Python Queue模塊的FIFO隊列先進先出。  class queue.Queue(maxsize)
2、LIFO類似於堆，即先進后出。             class queue.LifoQueue(maxsize)
3、還有一種是優先級隊列級別越低越先出來。   class queue.PriorityQueue(maxsize)

此包中的常用方法(q = Queue.Queue()):
q.qsize() 返回隊列的大小
q.empty() 如果隊列為空，返回True,反之False
q.full() 如果隊列滿了，返回True,反之False
q.full 與 maxsize 大小對應
q.get([block[, timeout]]) 獲取隊列，timeout等待時間
q.get_nowait() 相當q.get(False)
非阻塞 q.put(item) 寫入隊列，timeout等待時間
q.put_nowait(item) 相當q.put(item, False)
q.task_done() 在完成一項工作之后，q.task_done() 函數向任務已經完成的隊列發送一個信號
q.join() 實際上意味着等到隊列為空，再執行別的操作

 

例子集合：

#例子1：    
import threading,queue
from time import sleep
from random import randint
class Production(threading.Thread):
    def run(self):
        while True:
            r=randint(0,100)
            q.put(r)
            print("生產出來%s號包子"%r)
            sleep(1)
class Proces(threading.Thread):
    def run(self):
        while True:
            re=q.get()
            print("吃掉%s號包子"%re)
if __name__=="__main__":
    q=queue.Queue(10)
    threads=[Production(),Production(),Production(),Proces()]
    for t in threads:
        t.start()

#例子2
import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
def Consumer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        print(data)
        print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
p1.start()
c1.start()

#例子3
#實現一個線程不斷生成一個隨機數到一個隊列中(考慮使用Queue這個模塊)
# 實現一個線程從上面的隊列里面不斷的取出奇數
# 實現另外一個線程從上面的隊列里面不斷取出偶數

import random,threading,time
from queue import Queue
#Producer thread
class Producer(threading.Thread):
  def __init__(self, t_name, queue):
    threading.Thread.__init__(self,name=t_name)
    self.data=queue
  def run(self):
    for i in range(10):  #隨機產生10個數字 ，可以修改為任意大小
      randomnum=random.randint(1,99)
      print ("%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum))
      self.data.put(randomnum) #將數據依次存入隊列
      time.sleep(1)
    print ("%s: %s finished!" %(time.ctime(), self.getName()))

#Consumer thread
class Consumer_even(threading.Thread):
  def __init__(self,t_name,queue):
    threading.Thread.__init__(self,name=t_name)
    self.data=queue
  def run(self):
    while 1:
      try:
        val_even = self.data.get(1,5) #get(self, block=True, timeout=None) ,1就是阻塞等待,5是超時5秒
        if val_even%2==0:
          print ("%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(),self.getName(),val_even))
          time.sleep(2)
        else:
          self.data.put(val_even)
          time.sleep(2)
      except:   #等待輸入，超過5秒 就報異常
        print ("%s: %s finished!" %(time.ctime(),self.getName()))
        break
class Consumer_odd(threading.Thread):
  def __init__(self,t_name,queue):
    threading.Thread.__init__(self, name=t_name)
    self.data=queue
  def run(self):
    while 1:
      try:
        val_odd = self.data.get(1,5)
        if val_odd%2!=0:
          print ("%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_odd))
          time.sleep(2)
        else:
          self.data.put(val_odd)
          time.sleep(2)
      except:
        print ("%s: %s finished!" % (time.ctime(), self.getName()))
        break
#Main thread
def main():
  queue = Queue()
  producer = Producer('Pro.', queue)
  consumer_even = Consumer_even('Con_even.', queue)
  consumer_odd = Consumer_odd('Con_odd.',queue)
  producer.start()
  consumer_even.start()
  consumer_odd.start()
  producer.join()
  consumer_even.join()
  consumer_odd.join()
  print ('All threads terminate!')

if __name__ == '__main__':
  main()

#注意：列表是線程不安全的
#例子4
import threading,time

li=[1,2,3,4,5]

def pri():
    while li:
        a=li[-1]
        print(a)
        time.sleep(1)
        try:
            li.remove(a)
        except:
            print('----',a)

t1=threading.Thread(target=pri,args=())
t1.start()

 

t2=threading.Thread(target=pri,args=())
t2.start()