當(dāng)前位置：首頁 > 编程语言 > python >内容正文

python

[Python 多线程] Semaphore、BounedeSemaphore (十二)

發(fā)布時間：2025/6/17 python 24 豆豆

生活随笔收集整理的這篇文章主要介紹了 [Python 多线程] Semaphore、BounedeSemaphore (十二) 小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

?Semaphore

信號量，信號量對象內(nèi)部維護一個倒計數(shù)器，每一次acquire都會減1，當(dāng)acquire方法發(fā)現(xiàn)計數(shù)為0就阻塞請求的線程，直到其它線程對信號量release后，計數(shù)大于0，恢復(fù)阻塞的線程。

方法：

Semaphore(value=1)? ? ? ? ? ? ? ? ? ? ? ? ? ? 構(gòu)造方法。value小于0，拋ValueError異常。默認(rèn)為1。

acquire(blocking=True,timeout=None)? 獲取信號量，計數(shù)器減1，獲取成功返回True。

release()? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?釋放信號量，計數(shù)器加1。

計數(shù)器永遠不會低于0，因為acquire的時候，發(fā)現(xiàn)是0，都會被阻塞。

?舉例：

圖書館有三本書，三本都被借走(acquire)之后，其他人想看，就得等別人還回來(阻塞)，有人還回來(release)一本后，就有一個人可以拿到這本書，其他人仍然得等歸還。

#Semaphore 信號量，借還 import threading,logging,time DATEFMT="%H:%M:%S" FORMAT = "[%(asctime)s]\t [%(threadName)s,%(thread)d] %(message)s" logging.basicConfig(level=logging.INFO,format=FORMAT,datefmt=DATEFMT)def work(s:threading.Semaphore):logging.info('in sub thread')logging.info(s.acquire())logging.info('sub thread oevr')s = threading.Semaphore(3) logging.info(s.acquire()) logging.info(s.acquire()) logging.info(s.acquire())threading.Thread(target=work,args=(s,)).start() time.sleep(2)logging.info(s.acquire(False)) #不阻塞 logging.info((s.acquire(timeout=3))) #3秒超時會阻塞logging.info('release') s.release()運行結(jié)果： [08:48:43] [MainThread,8840] True [08:48:43] [MainThread,8840] True [08:48:43] [MainThread,8840] True [08:48:43] [Thread-1,6212] in sub thread [08:48:45] [MainThread,8840] False [08:48:48] [MainThread,8840] False [08:48:48] [MainThread,8840] release [08:48:48] [Thread-1,6212] True [08:48:48] [Thread-1,6212] sub thread oevr

　　這個例子只起了一個線程，如果多起幾個，當(dāng)release還回來的數(shù)小于阻塞的線程數(shù)時，程序就會一直處于阻塞狀態(tài)，直到全部relase。

?應(yīng)用舉例：

因為資源有限，且開啟一個連接成本高，所以，使用連接池。

一個簡單的連接池(例子)：

連接池應(yīng)該有容量(value總數(shù))，也應(yīng)該工廠方法可以獲取連接，能夠把不用的連接歸還，供其他使用者使用。

#一個簡單的連接池 import threading,logging,time DATEFMT="%H:%M:%S" FORMAT = "[%(asctime)s]\t [%(threadName)s,%(thread)d] %(message)s" logging.basicConfig(level=logging.INFO,format=FORMAT,datefmt=DATEFMT)class Conn:def __init__(self,name):self.name = nameclass Pool:def __init__(self,count=3):self.count = count#連接池容器self.pool = [self._connect('conn-{}'.format(x)) for x in range(self.count)]def _connect(self,conn_name):return Conn(conn_name)def get_conn(self):# if len(self.pool) > 0:return self.pool.pop() #從尾部彈出一個def return_conn(self,conn:Conn):self.pool.append(conn)pool = Pool(3) print(pool.pool) pool.get_conn() pool.get_conn() pool.get_conn() pool.get_conn() #第4個print('End Main')運行結(jié)果： [<__main__.Conn object at 0x00000211BBEBC160>, <__main__.Conn object at 0x00000211BBEBC1D0>, <__main__.Conn object at 0x00000211BBEBC240>] Traceback (most recent call last):File "C:/python/test.py", line 34, in <module>pool.get_conn()File "C:/python/test.py", line 24, in get_connreturn self.pool.pop() #從尾部彈出一個 IndexError: pop from empty list

　　當(dāng)連接池中已經(jīng)沒有可用連接時，再獲取就會拋異常 IndexError：pop from empty list。

那就加個判斷，只在池中連接數(shù)量大于0的時候才可以獲取連接：

#修改get_conn函數(shù)def get_conn(self):if len(self.pool) > 0:return self.pool.pop() #從尾部彈出一個

　　這樣在連接池為空時，就不會拋異常了。

這個連接池的例子如果使用多線程，這個get_conn()方法是線程不安全的，有可能其它線程看到池中還有一個連接，正準(zhǔn)備獲取，其它線程也看到了，也準(zhǔn)備獲取連接，就會拋異常。再或者，都在向池中加連接的時候，也可能會多加。

這個問題可以用鎖Lock來解決，在獲取連接和加連接時，加鎖解鎖；也可以使用semaphore信號量來解決。

使用信號量對上例進行修改：

#使用semaphore信號量修改連接池 import threading,logging,time,random DATEFMT="%H:%M:%S" FORMAT = "[%(asctime)s]\t [%(threadName)s,%(thread)d] %(message)s" logging.basicConfig(level=logging.INFO,format=FORMAT,datefmt=DATEFMT)class Conn:def __init__(self,name):self.name = namedef __repr__(self):return self.nameclass Pool:def __init__(self,count=3):self.count = count#連接池容器self.pool = [self._connect('conn-{}'.format(x)) for x in range(self.count)]self.semaphore = threading.Semaphore(self.count)def _connect(self,conn_name):#返回一個連接名return Conn(conn_name)def get_conn(self):#從池中拿走一個連接# if len(self.pool) > 0:self.semaphore.acquire(timeout=5) #-1,獲取連接,最大5秒超時時間,與后面隨機秒數(shù)相對應(yīng)data = self.pool.pop() #從尾部彈出一個return datadef return_conn(self,conn:Conn):#向池中添加一個連接self.pool.append(conn)self.semaphore.release() # 先加入池中再信號量+1return len(self.pool)pool = Pool(3)def worker(pool:Pool):conn = pool.get_conn()logging.info(conn)#模擬使用了資源一段時間(隨機1-4秒),然后歸還threading.Event().wait(timeout=random.randint(1,4))pool.return_conn(conn)for i in range(6):threading.Thread(target=worker,name="worker-{}".format(i),args=(pool,)).start()print('End Main')運行結(jié)果： [10:34:12] [worker-0,5264] conn-2 [10:34:12] [worker-1,7420] conn-1 [10:34:12] [worker-2,2612] conn-0 End Main [10:34:13] [worker-3,3972] conn-1 #歸還以后又可以獲取連接 [10:34:14] [worker-4,8172] conn-2 [10:34:15] [worker-5,11192] conn-1

　　上例中模擬獲取連接以后使用了1-4秒鐘，沒有拿到資源的最多阻塞5秒鐘，當(dāng)連接使用結(jié)束歸還后，阻塞的線程就又重新獲取到連接。

問題：

1) 沒有使用信號量就release的情況：

import threadings = threading.Semaphore(3) print(s.__dict__)def work(s:threading.Semaphore):s.release()for i in range(3):threading.Thread(target=work,args=(s,)).start()print(s.__dict__)運行結(jié)果： {'_cond': <Condition(<unlocked _thread.lock object at 0x00000219202973A0>, 0)>, '_value': 3} {'_cond': <Condition(<unlocked _thread.lock object at 0x00000219202973A0>, 0)>, '_value': 2} {'_cond': <Condition(<unlocked _thread.lock object at 0x00000219202973A0>, 0)>, '_value': 3} {'_cond': <Condition(<unlocked _thread.lock object at 0x00000219202973A0>, 0)>, '_value': 4} {'_cond': <Condition(<unlocked _thread.lock object at 0x00000219202973A0>, 0)>, '_value': 5}

　　沒有acquire信號量時，就release的情況，結(jié)果導(dǎo)致了信號量的內(nèi)置倒計數(shù)器的值增加，這樣就超出了最大值。

解決辦法：

使用BoundedSemaphore類：

BoundedSemaphore，繼承自Semaphore類。邊界綁定，有界的信號量，不允許使用release超過初始值的范圍，否則，拋ValueError異常。

#BoundedSemaphore邊界綁定 import threadings = threading.BoundedSemaphore(3) print(s.__dict__)s.acquire() print(s.__dict__)def work(s:threading.BoundedSemaphore):s.release()for i in range(3):threading.Thread(target=work,args=(s,)).start()print(s.__dict__)運行結(jié)果： {'_value': 3, '_cond': <Condition(<unlocked _thread.lock object at 0x000001A42DDF73A0>, 0)>, '_initial_value': 3} {'_value': 2, '_cond': <Condition(<unlocked _thread.lock object at 0x000001A42DDF73A0>, 0)>, '_initial_value': 3} {'_value': 3, '_cond': <Condition(<unlocked _thread.lock object at 0x000001A42DDF73A0>, 0)>, '_initial_value': 3} {'_value': 3, '_cond': <Condition(<unlocked _thread.lock object at 0x000001A42DDF73A0>, 0)>, '_initial_value': 3} {'_value': 3, '_cond': <Condition(<unlocked _thread.lock object at 0x000001A42DDF73A0>, 0)>, '_initial_value': 3} Exception in thread Thread-2: Traceback (most recent call last):File "C:/python/test.py", line 11, in works.release() ValueError: Semaphore released too many times

　　使用BoundedSemaphore就可以控制不會多歸還。

轉(zhuǎn)載于:https://www.cnblogs.com/i-honey/p/8078518.html

總結(jié)

以上是生活随笔為你收集整理的[Python 多线程] Semaphore、BounedeSemaphore (十二)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。

上一篇： selenium webdrive 默认
下一篇： C# ManualResetEvent