threading庫：Python線程鎖與釋放鎖

目錄

控制資源訪問

判斷是否有另一個線程請求鎖

with lock

同步線程

Condition

屏障（barrier）

有限資源的并發訪問

隱藏資源

控制資源訪問

前文提到threading庫在多線程時，對同一資源的訪問容易導致破壞與丟失數據。為了保證安全的訪問一個資源對象，我們需要創建鎖。

示例如下：

import threading import time class AddThread(): def __init__(self, start=0): self.lock = threading.Lock() self.value = start def increment(self): print("Wait Lock") self.lock.acquire() try: print("Acquire Lock") self.value += 1 print(self.value) finally: self.lock.release() def worker(a): time.sleep(1) a.increment() addThread = AddThread() for i in range(3): t = threading.Thread(target=worker, args=(addThread,)) t.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

運行之后，效果如下：

acquire()會通過鎖進行阻塞其他線程執行中間段，release()釋放鎖，可以看到，基本都是獲得鎖之后才執行。避免了多個線程同時改變其資源對象，不會造成混亂。

判斷是否有另一個線程請求鎖

要確定是否有另一個線程請求鎖而不影響當前的線程，可以設置acquire()的參數blocking=False。

示例如下：

import threading import time def worker2(lock): print("worker2 Wait Lock") while True: lock.acquire() try: print("Holding") time.sleep(0.5) finally: print("not Holding") lock.release() time.sleep(0.5) def worker1(lock): print("worker1 Wait Lock") num_acquire = 0 value = 0 while num_acquire < 3: time.sleep(0.5) have_it = lock.acquire(blocking=False) try: value += 1 print(value) print("Acquire Lock") if have_it: num_acquire += 1 finally: print("release Lock") if have_it: lock.release() lock = threading.Lock() word2Thread = threading.Thread( target=worker2, name='work2', args=(lock,) ) word2Thread.start() word1Thread = threading.Thread( target=worker1, name='work1', args=(lock,) ) word1Thread.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

運行之后，效果如下：

這里，我們需要迭代很多次，work1才能獲取3次鎖。但是嘗試了很8次。

with lock

前文，我們通過lock.acquire()與lock.release()實現了鎖的獲取與釋放，但其實我們Python還給我們提供了一個更簡單的語法，通過with lock來獲取與釋放鎖。

示例如下：

import threading import time class AddThread(): def __init__(self, start=0): self.lock = threading.Lock() self.value = start def increment(self): print("Wait Lock") with self.lock: print("lock acquire") self.value += 1 print(self.value) print("lock release") def worker(a): time.sleep(1) a.increment() addThread = AddThread() for i in range(3): t = threading.Thread(target=worker, args=(addThread,)) t.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

這里，我們只是將最上面的例子改變了一下。效果如下：

需要注意的是，正常的Lock對象不能請求多次，即使是由同一個線程請求也不例外。如果同一個調用鏈中的多個函數訪問一個鎖，則會發生意外。如果期望在同一個線程的不同代碼需要重新獲得鎖，那么這種情況下使用RLock。

同步線程

Condition

在實際的操作中，我們還可以使用Condition對象來同步線程。由于Condition使用了一個Lock，所以它可以綁定到一個共享資源，允許多個線程等待資源的更新。

示例如下：

import threading import time def consumer(cond): print("waitCon") with cond: cond.wait() print('獲取更新的資源') def producer(cond): print("worker") with cond: print('更新資源') cond.notifyAll() cond = threading.Condition() t1 = threading.Thread(name='t1', target=consumer, args=(cond,)) t2 = threading.Thread(name='t2', target=consumer, args=(cond,)) t3 = threading.Thread(name='t3', target=producer, args=(cond,)) t1.start() time.sleep(0.2) t2.start() time.sleep(0.2) t3.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

運行之后，效果如下：

這里，我們通過producer線程處理完成之后調用notifyAll()，consumer等線程等到了它的更新，可以類比為觀察者模式。這里是，當一個線程用完資源之后時,則會自動通知依賴它的所有線程。

屏障（barrier）

屏障是另一種線程的同步機制。barrier會建立一個控制點，所有參與的線程會在這里阻塞，直到所有這些參與方都到達這一點。采用這種方法，線程可以單獨啟動然后暫停，直到所有線程都準備好了才可以繼續。

示例如下：

import threading import time def worker(barrier): print(threading.current_thread().getName(), "worker") worker_id = barrier.wait() print(threading.current_thread().getName(), worker_id) threads = [] barrier = threading.Barrier(3) for i in range(3): threads.append( threading.Thread( name="t" + str(i), target=worker, args=(barrier,) ) ) for t in threads: print(t.name, 'starting') t.start() time.sleep(0.1) for t in threads: t.join()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

運行之后，效果如下：

從控制臺的輸出會發發現，barrier.wait()會阻塞線程，直到所有線程被創建后，才同時釋放越過這個控制點繼續執行。wait()的返回值指示了釋放的參與線程數，可以用來限制一些線程做清理資源等動作。

當然屏障Barrier還有一個abort()方法，該方法可以使所有等待線程接收一個BroKenBarrierError。如果線程在wait()上被阻塞而停止處理，會產生這個異常，通過except可以完成清理工作。

有限資源的并發訪問

除了多線程可能訪問同一個資源之外，有時候為了性能，我們也會限制多線程訪問同一個資源的數量。例如，線程池支持同時連接，但數據可能是固定的，或者一個網絡APP提供的并發下載數支持固定數目。這些連接就可以使用Semaphore來管理。

示例如下：

import threading import time class WorkerThread(threading.Thread): def __init__(self): super(WorkerThread, self).__init__() self.lock = threading.Lock() self.value = 0 def increment(self): with self.lock: self.value += 1 print(self.value) def worker(s, pool): with s: print(threading.current_thread().getName()) pool.increment() time.sleep(1) pool.increment() pool = WorkerThread() s = threading.Semaphore(2) for i in range(5): t = threading.Thread( name="t" + str(i), target=worker, args=(s, pool,) ) t.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

運行之后，效果如下：

從圖片雖然能看所有輸出，但無法看到其停頓的事件。讀者自己運行會發現，每次頂多只有兩個線程在工作，是因為我們設置了threading.Semaphore(2)。

隱藏資源

在實際的項目中，有些資源需要鎖定以便于多個線程使用，而另外一些資源則需要保護，以使它們對并非使這些資源的所有者的線程隱藏。

local()函數會創建一個對象，它能夠隱藏值，使其在不同的線程中無法被看到。示例如下：

import threading import random def show_data(data): try: result = data.value except AttributeError: print(threading.current_thread().getName(), "No value") else: print(threading.current_thread().getName(), "value=", result) def worker(data): show_data(data) data.value = random.randint(1, 100) show_data(data) local_data = threading.local() show_data(local_data) local_data.value = 1000 show_data(local_data) for i in range(2): t = threading.Thread( name="t" + str(i), target=worker, args=(local_data,) ) t.start()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

運行之后，效果如下：

這里local_data.value對所有線程都不可見，除非在某個線程中設置了這個屬性，這個線程才能看到它。

Python 任務調度

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。