python類庫32[多線程同步Lock+RLock+Semaphore+Event]

 

多線程基礎：python類庫32[多線程同步]

 

一 多線程同步

由於CPython的python解釋器在單線程模式下執行，所以導致python的多線程在很多的時候並不能很好地發揮多核cpu的資源。大部分情況都推薦使用多進程。

python的多線程的同步與其他語言基本相同，主要包含：

Lock & RLock ：用來確保多線程多共享資源的訪問。

Semaphore ： 用來確保一定資源多線程訪問時的上限，例如資源池。  

Event : 是最簡單的線程間通信的方式，一個線程可以發送信號，其他的線程接收到信號后執行操作。  

 

二 實例

1）Lock & RLock

Lock對象的狀態可以為locked和unlocked，
使用acquire()設置為locked狀態；
使用release()設置為unlocked狀態。

如果當前的狀態為unlocked，則acquire()會將狀態改為locked然后立即返回。當狀態為locked的時候，acquire()將被阻塞直到另一個線程中調用release()來將狀態改為unlocked，然后acquire()才可以再次將狀態置為locked。 

Lock.acquire(blocking=True, timeout=-1),blocking參數表示是否阻塞當前線程等待，timeout表示阻塞時的等待時間 。如果成功地獲得lock，則acquire()函數返回True，否則返回False，timeout超時時如果還沒有獲得lock仍然返回False。

 

實例：(確保只有一個線程可以訪問共享資源)

 
          import threading 
          
 
          import time 
          
  
          
num = 0 
          
lock = threading.Lock() 
          
  
          
 
          def func(st): 
          
     
          global num 
          
     
          print (threading.currentThread().getName() +  
          ' 
           try to acquire the lock 
          ') 
          
     
          if lock.acquire(): 
          
         
          print (threading.currentThread().getName() +  
          ' 
           acquire the lock. 
          ' ) 
          
         
          print (threading.currentThread().getName() + 
          " 
           :%s 
          " % str(num) ) 
          
        num += 1 
          
        time.sleep(st) 
          
         
          print (threading.currentThread().getName() +  
          ' 
           release the lock. 
          '  )         
          
        lock.release() 
          
  
          
t1 = threading.Thread(target=func, args=(8,)) 
          
t2 = threading.Thread(target=func, args=(4,)) 
          
t3 = threading.Thread(target=func, args=(2,)) 
          
t1.start() 
          
t2.start() 
          
t3.start() 
         

 

 

結果：


RLock與Lock的區別是：RLock中除了狀態locked和unlocked外還記錄了當前lock的owner和遞歸層數，使得RLock可以被同一個線程多次acquire()。

 

2）Semaphore 

Semaphore管理一個內置的計數器，

每當調用acquire()時內置計數器-1；

調用release() 時內置計數器+1；

計數器不能小於0；當計數器為0時，acquire()將阻塞線程直到其他線程調用release()。 

 

實例：(同時只有2個線程可以獲得semaphore,即可以限制最大連接數為2)： 

 
          import threading 
          
 
          import time  
          
 
          
semaphore = threading.Semaphore(2) 
          
  
          
 
          def func(): 
          
     
          if semaphore.acquire(): 
          
         
          for i  
          in range(5): 
          
           
          print (threading.currentThread().getName() +  
          ' 
           get semaphore 
          ') 
          
        semaphore.release() 
          
         
          print (threading.currentThread().getName() +  
          ' 
           release semaphore 
          ') 
          
         
          
         
          
 
          for i  
          in range(4): 
          
  t1 = threading.Thread(target=func) 
          
  t1.start() 
         

 

 

結果：

 

 

3) Event 

Event內部包含了一個標志位，初始的時候為false。
可以使用使用set()來將其設置為true；
或者使用clear()將其從新設置為false；
可以使用is_set()來檢查標志位的狀態；
另一個最重要的函數就是wait(timeout=None)，用來阻塞當前線程，直到event的內部標志位被設置為true或者timeout超時。如果內部標志位為true則wait()函數理解返回。

 

實例： (線程間相互通信)

 
             import logging 
             
 
             import threading 
             
 
             import time 
             
 
             
logging.basicConfig(level=logging.DEBUG, 
             
format= 
             " 
             (%(threadName)-10s : %(message)s 
             ", 
             
) 
             
 
             
 
             def wait_for_event_timeout(e, t): 
             
     
             """ 
             Wait t seconds and then timeout 
             """ 
             
     
             while  
             not e.isSet(): 
             
      logging.debug( 
             " 
             wait_for_event_timeout starting 
             ") 
             
      event_is_set = e.wait(t) 
             
      logging.debug( 
             " 
             event set: %s 
             " % event_is_set) 
             
     
             if event_is_set: 
             
      logging.debug( 
             " 
             processing event 
             ") 
             
     
             else: 
             
      logging.debug( 
             " 
             doing other work 
             ") 
             
       
             
e = threading.Event() 
             
t2 = threading.Thread(name= 
             " 
             nonblock 
             ", 
             
target=wait_for_event_timeout,args=(e, 2)) 
             
t2.start() 
             
logging.debug( 
             " 
             Waiting before calling Event.set() 
             ") 
             
time.sleep(7) 
             
e.set() 
             
logging.debug( 
             " 
             Event is set 
             ") 
            

 

運行結果：

  

 

三 其他

1） 線程局部變量

線程局部變量的值是跟線程相關的，區別與全局的變量。使用非常簡單如下：

mydata = threading.local()

mydata.x = 1

 

2）對Lock，semaphore，condition等使用with關鍵字代替手動調用acquire()和release()。 

 

完！