Python線程同步

線程執行

join與setDaemon

子線程在主線程運行結束后，會繼續執行完，如果給子線程設置為守護線程(setDaemon=True)，主線程運行結束子線程即結束；

如果join()線程，那么主線程會等待子線程執行完再執行。

 

import threading  2 import time  3 

def get_thread_a():  6     print("get thread A started")  7     time.sleep(3)  8     print("get thread A end")  9 

def get_thread_b(): 12     print("get thread B started") 13     time.sleep(5) 14     print("get thread B end") 15 

if  __name__ == "__main__": 18     thread_a = threading.Thread(target=get_thread_a) 19     thread_b = threading.Thread(target=get_thread_b) 20     start_time = time.time() 21  thread_b.setDaemon(True) 22  thread_a.start() 23  thread_b.start() 24  thread_a.join() 25     
    end_time = time.time() 27     print("execution time: {}".format(end_time - start_time))

 

thread_a是join，首先子線程thread_a執行，thread_b是守護線程，當主線程執行完后，thread_b不會再執行
執行結果如下：

get thread A started get thread B started get thread A end execution time: 3.003199815750122

線程同步

當線程間共享全局變量，多個線程對該變量執行不同的操作時，該變量最終的結果可能是不確定的(每次線程執行后的結果不同)，如：對count變量執行加減操作 ，count的值是不確定的，要想count的值是一個確定的需對線程執行的代碼段加鎖。

python對線程加鎖主要有Lock和Rlock模塊

Lock：

from threading import Lock

lock = Lock()
lock.acquire()
lock.release()

Lock有acquire()和release()方法，這兩個方法必須是成對出現的，acquire()后面必須release()后才能再acquire()，否則會造成死鎖

Rlock：

鑒於Lock可能會造成死鎖的情況，RLock(可重入鎖)對Lock進行了改進，RLock可以在同一個線程里面連續調用多次acquire()，但必須再執行相同次數的release()

from threading import RLock lock = RLock() lock.acquire() lock.acquire() lock.release() lock.release()

condition(條件變量)，線程在執行時，當滿足了特定的條件后，才可以訪問相關的數據

import threading def get_thread_a(condition): with condition: condition.wait() print("A : Hello B,that's ok") condition.notify() condition.wait() print("A : I'm fine,and you?") condition.notify() condition.wait() print("A : Nice to meet you") condition.notify() condition.wait() print("A : That's all for today") condition.notify() def get_thread_b(condition): with condition: print("B : Hi A, Let's start the conversation") condition.notify() condition.wait() print("B : How are you") condition.notify() condition.wait() print("B : I'm fine too") condition.notify() condition.wait() print("B : Nice to meet you,too") condition.notify() condition.wait() print("B : Oh,goodbye") if __name__ == "__main__": condition = threading.Condition() thread_a = threading.Thread(target=get_thread_a, args=(condition,)) thread_b = threading.Thread(target=get_thread_b, args=(condition,)) thread_a.start() thread_b.start()

Condition內部有一把鎖，默認是RLock，在調用wait()和notify()之前必須先調用acquire()獲取這個鎖，才能繼續執行；當wait()和notify()執行完后，需調用release()釋放這個鎖，在執行with condition時，會先執行acquire()，with結束時，執行了release()；所以condition有兩層鎖，最底層鎖在調用wait()時會釋放，同時會加一把鎖到等待隊列，等待notify()喚醒釋放鎖

wait() ：允許等待某個條件變量的通知，notify()可喚醒

notify()： 喚醒等待隊列wait()

執行結果：

B : Hi A, Let's start the conversation
A : Hello B,that's ok
B : How are you A : I'm fine,and you?
B : I'm fine too
A : Nice to meet you B : Nice to meet you,too A : That's all for today
B : Oh,goodbye

 

 

Semaphore(信號量)

 用於控制線程的並發數，如爬蟲中請求次數過於頻繁會被禁止ip，每次控制爬取網頁的線程數量可在一定程度上防止ip被禁；文件讀寫中，控制寫線程每次只有一個，讀線程可多個。

import time import threading def get_thread_a(semaphore,i): time.sleep(1) print("get thread : {}".format(i)) semaphore.release() def get_thread_b(semaphore): for i in range(10): semaphore.acquire() thread_a = threading.Thread(target=get_thread_a, args=(semaphore,i)) thread_a.start() if __name__ == "__main__": semaphore = threading.Semaphore(2) thread_b = threading.Thread(target=get_thread_b, args=(semaphore,)) thread_b.start()

上述示例了每隔1秒並發兩個線程執行的情況，當調用一次semaphore.acquire()時，Semaphore的數量就減1，直至Semaphore數量為0時被鎖上，當release()后Semaphore數量加1。Semaphore在本質上是調用的Condition，semaphore.acquire()在Semaphore的值為0的條件下會調用Condition.wait(), 否則將值減1，semaphore.release()會將Semaphore的值加1，並調用Condition.notify()

Semaphore源碼

def acquire(self, blocking=True, timeout=None): if not blocking and timeout is not None: raise ValueError("can't specify timeout for non-blocking acquire") rc = False endtime = None with self._cond: while self._value == 0: if not blocking: break
                if timeout is not None: if endtime is None: endtime = _time() + timeout else: timeout = endtime - _time() if timeout <= 0: break self._cond.wait(timeout) else: self._value -= 1 rc = True return rc def release(self): with self._cond: self._value += 1 self._cond.notify()