日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當前位置:首頁 > 科技  > 軟件

源碼解密協程隊列和線程隊列的實現原理

來源: 責編: 時間:2023-12-05 09:24:26 247觀看
導讀本次來聊一聊 Python 的隊列,首先隊列是一種特殊的線性表,具有先進先出(FIFO)的特性,這意味著元素的入隊順序和出隊順序是一致的。隊列通常用于存儲需要按順序處理的數據,例如任務調度。當然隊列最常見的一個應用場景就是解

本次來聊一聊 Python 的隊列,首先隊列是一種特殊的線性表,具有先進先出(FIFO)的特性,這意味著元素的入隊順序和出隊順序是一致的。BHG28資訊網——每日最新資訊28at.com

圖片BHG28資訊網——每日最新資訊28at.com

隊列通常用于存儲需要按順序處理的數據,例如任務調度。當然隊列最常見的一個應用場景就是解耦,一個線程不停地生產數據,放到隊列里,另一個線程從隊列中取數據進行消費。BHG28資訊網——每日最新資訊28at.com

而 Python 也提供了隊列,分別是協程隊列和線程隊列。BHG28資訊網——每日最新資訊28at.com

import asyncioimport queue# 協程隊列coroutine_queue = asyncio.Queue()# 線程隊列threading_queue = queue.Queue()

如果你的程序基于 asyncio,那么應該使用協程隊列,如果你的程序采用了多線程,那么應該使用線程隊列。BHG28資訊網——每日最新資訊28at.com

下面我們來看一看這兩種隊列的 API,以及底層實現原理。BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

協程隊列BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

協程隊列的具體實現由 asyncio 提供,以下是它的一些用法。BHG28資訊網——每日最新資訊28at.com

import asyncioasync def main():    # 創建隊列時可以指定能夠存儲的最大元素個數    # 不指定則沒有容量限制    queue = asyncio.Queue(maxsize=20)    # 返回容量    print(queue.maxsize)    """    20    """    # 添加元素,如果隊列滿了會阻塞,直到有剩余空間    await queue.put(111)    # 添加元素,如果隊列滿了會拋異常    # 因為不需要阻塞等待,所以 put_nowait 不是協程函數    queue.put_nowait(222)    # 隊列是否已滿    print(queue.full())    """    False    """    # 返回隊列內部的元素個數    print(queue.qsize())    """    2    """    # 從隊列中獲取元素,如果隊列為空,會阻塞,直到隊列中有可用元素    print(await queue.get())    """    111    """    # 從隊列中獲取元素,如果隊列為空,會拋異常    # 因為不需要阻塞等待,所以 put_nowait 不是協程函數    print(queue.get_nowait())    """    222    """    # 隊列是否為空    print(queue.empty())    """    True    """asyncio.run(main())

所以協程隊列的 API 很簡單,我們再羅列一下:BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

然后,協程隊列還有兩個 API,需要單獨說明,分別是 task_done() 和 join()。BHG28資訊網——每日最新資訊28at.com

首先在協程隊列內部有一個 _unfinished_tasks 屬性,初始值為 0,每當往隊列添加一個元素時,該屬性的值就會自增 1。但是從隊列取出元素時,該屬性不會自動減 1,需要手動調用 task_done() 方法。BHG28資訊網——每日最新資訊28at.com

所以 _unfinished_tasks 記錄了隊列中有多少個任務數據需要處理,每來一個自動加 1,但取走一個不會自動減 1,而是需要 task_done 來實現。BHG28資訊網——每日最新資訊28at.com

然后 join() 的作用是,當 _unfinished_tasks 不為 0 的時候,await queue.join() 會阻塞,直到為 0。BHG28資訊網——每日最新資訊28at.com

import asyncioasync def consumer(queue, n):    print(f"consumer{n} 開始消費")    await asyncio.sleep(3)    await queue.get()    # 獲取數據后,調用 task_done    queue.task_done()    print(f"consumer{n} 消費完畢")async def main():    queue = asyncio.Queue()    await queue.put(123)    await queue.put(456)    await queue.put(789)    # 隊列里面有三個數據,開啟三個消費者去消費    await asyncio.gather(        consumer(queue, 1),        consumer(queue, 2),        consumer(queue, 3),    )    # 這里會陷入阻塞,直到 _unfinished_tasks 變為 0    await queue.join()    print("main 解除阻塞")asyncio.run(main())"""consumer1 開始消費consumer2 開始消費consumer3 開始消費consumer1 消費完畢consumer2 消費完畢consumer3 消費完畢main 解除阻塞"""

還是比較簡單的,然后我們來看一下協程隊列的具體實現細節。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

首先協程隊列內部有一個 _queue 屬性,它是一個雙端隊列,負責保存具體的元素。因為要保證兩端的操作都是高效的,所以采用雙端隊列實現。BHG28資訊網——每日最新資訊28at.com

然后是 _getters 和 _putters 兩個屬性,它們是做什么的呢?在隊列滿了的時候,協程往隊列添加元素時會陷入阻塞,等到隊列有剩余空間時會解除阻塞。同理,在隊列為空時,協程從隊列獲取元素時會陷入阻塞,等到隊列有可用元素時會解除阻塞。BHG28資訊網——每日最新資訊28at.com

那么這個阻塞等待,以及自動喚醒并解除阻塞是怎么實現的呢?在介紹鎖和信號量的時候,我們分析過整個實現過程,協程隊列與之類似。BHG28資訊網——每日最新資訊28at.com

假設協程從隊列獲取元素,但是隊列為空,于是會創建一個 Future 對象,并保存起來,當前保存的地方就是 _getters,它也是雙端隊列。然后 await future,此時就會陷入阻塞,當其它協程往隊列中添加元素時,會將 _getters 里面的 future 彈出,設置結果集。因此 await future 的協程就會解除阻塞,因為隊列有可用元素了。BHG28資訊網——每日最新資訊28at.com

同理,協程往隊列添加元素也是如此,如果隊列滿了,同樣創建一個 Future 對象,并保存起來,當前保存的地方就是 _putters。然后 await future,陷入阻塞,當其它協程從隊列中取出元素,會將 _putters 里面的 future 彈出,設置結果集。因此 await future 的協程就會解除阻塞,因為隊列有可用空間了。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

三個內部調用的方法,_get 方法負責從隊列的頭部彈出元素,_put 方法負責從隊列的尾部追加元素,比較簡單。然后是 _wakeup_next 方法,它負責喚醒阻塞的協程。參數 waiters 要么是 _getters,要么是 _putters,從里面彈出一個 future,設置結果集,讓對應的協程解除阻塞。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

  • qsize() 負責返回隊列的元素個數;
  • maxsize 負責返回隊列的容量;
  • empty() 負責判斷隊列是否為空;
  • full() 負責判斷隊列是否已滿,如果容量小于等于 0,那么表示容量無限,隊列永遠不會滿。否則判斷元素個數是否大于等于容量;

圖片圖片BHG28資訊網——每日最新資訊28at.com

然后看看 put_nowait 和 get_nowait,首先是 put_nowait,往隊列添加元素。BHG28資訊網——每日最新資訊28at.com

如果添加時發現隊列已滿,那么拋出異常。如果未滿,則調用 _put 方法往 _queue 里面添加元素,因為元素的實際存儲是由 self._queue 這個雙端隊列負責的。BHG28資訊網——每日最新資訊28at.com

添加完畢后,將 _unfinished_task 加 1。最后從 _getters 里面彈出 future,設置結果集,讓因獲取不到元素而陷入阻塞的協程解除阻塞(同時會將添加的元素取走)。BHG28資訊網——每日最新資訊28at.com

get_nowait 的邏輯也很簡單,如果隊列為空,直接拋異常。如果不為空,則調用 _get 方法從隊列中彈出元素。最后從 _putters 里面彈出 future,設置結果集,讓因隊列已滿、無法添加元素而陷入阻塞的協程解除阻塞(同時會將元素添加進隊列)。BHG28資訊網——每日最新資訊28at.com

再來看看 put 方法的實現細節:BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

結果和我們之前分析的一樣,只是源碼內部多做了一些異常檢測。再來看看 get 方法,它的實現細節和 put 是類似的。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

比較簡單,還是沒什么難度的,最后再來看看 task_done 和 join 兩個方法。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

協程隊列里面使用了 asyncio.Event,它表示事件,如果事件對象沒有調用 set 方法設置標志位,那么調用 wait 方法時會陷入阻塞。當事件對象調用 set 方法時,wait 會解除阻塞。BHG28資訊網——每日最新資訊28at.com

所以協程隊列的 join 方法的邏輯就是,當 _unfinished_tasks 大于 0 時,調用事件對象的 wait 方法陷入阻塞。BHG28資訊網——每日最新資訊28at.com

而 task_done 方法的作用就是將 _unfinished_tasks 減 1,當它的值屬性為 0 時,調用事件對象的 set 方法,讓 join 解除阻塞。BHG28資訊網——每日最新資訊28at.com

以上就是整個協程隊列的實現細節,具體的元素存儲是由 collections.deque 來承載的。并在隊列已滿或者為空時,通過 Future 對象來實現阻塞等待和自動喚醒。BHG28資訊網——每日最新資訊28at.com

另外除了先進先出隊列之外,還有先進后出隊列,一般稱為 LIFO 隊列,它的效果類似于棧。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

這個沒什么好說的,因為是先進后出,所以添加和彈出都在同一端,直接使用列表實現即可。并且由于 LifoQueue 繼承 Queue,所以它的 API 和普通的協程隊列是一樣的。BHG28資訊網——每日最新資訊28at.com

除了先進先出隊列,還有一個優先隊列。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

它的 API 和普通的協程隊列也是一致的,只不過優先隊列在添加元素時,需要指定一個優先級:(優先級, 元素),優先級的值越低,表示優先級越高。然后在內部,會按照優先級的高低,維護一個小根堆,堆頂元素便是優先級最高的元素。BHG28資訊網——每日最新資訊28at.com

這幾個隊列具體使用哪一種,則取決于具體的業務場景。BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

線程隊列

BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

說完了協程隊列,再來看看線程隊列,它們的 API 是類似的,但實現細節則不同。因為操作系統感知不到協程,所以協程隊列的阻塞等待是基于 Future 實現的,而線程隊列的阻塞等待是基于條件變量(和互斥鎖)實現的。BHG28資訊網——每日最新資訊28at.com

還是先來看看線程隊列的一些 API,和協程隊列是類似的。BHG28資訊網——每日最新資訊28at.com

from queue import Queue# 可以指定一個 maxsize 參數,表示隊列的容量# 默認為 0,表示隊列的容量無限queue = Queue(maxsize=20)# 查看容量print(queue.maxsize)"""20"""# 查看隊列的元素個數print(queue.qsize())"""0"""# 判斷隊列是否已滿print(queue.full())"""False"""# 判斷隊列是否為空print(queue.empty())"""True"""# 往隊列中添加元素# block 參數表示是否阻塞,默認為 True,當隊列已滿時,線程會阻塞# timeout 表示超時時間,默認為 None,表示會無限等待# 當然也可以給 timeout 傳一個具體的值# 如果在規定時間內,沒有將元素放入隊列,那么拋異常queue.put(123, block=True, timeout=None)# 也是往隊列中添加元素,但是當隊列已滿時,會直接拋異常# put_nowait(item) 本質上就是 put(item, block=False)queue.put_nowait(456)# 從隊列中取出元素# 同樣可以傳遞 block 和 timeout 參數# block 默認為 True,當隊列為空時會陷入阻塞# timeout 默認為 None,表示會無限等待print(queue.get(block=True, timeout=None))"""123"""# 也是從隊列中取出元素,但是當隊列為空時,會直接拋異常# get_nowait() 本質上就是 get(block=False)print(queue.get_nowait())"""456"""# task_done(),將 unfinished_tasks 屬性的值減 1print(queue.unfinished_tasks)  """2"""queue.task_done()queue.task_done()print(queue.unfinished_tasks)"""0"""# join(),當 unfinished_tasks 不為 0 時,陷入阻塞queue.join()

API 和協程隊列是相似的,我們羅列一下:BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

線程隊列的具體使用我們已經知道了,下面來看看它的具體實現。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

線程隊列的內部依舊使用雙端隊列進行元素存儲,并且還使用了一個互斥鎖和三個條件變量。BHG28資訊網——每日最新資訊28at.com

為了保證數據的一致性和線程安全,當隊列在多線程環境中被修改(比如添加或刪除元素)時,需要使用互斥鎖。任何需要修改隊列的操作都必須在獲取到互斥鎖之后進行,以防止多個線程同時對隊列進行修改,否則會導致數據不一致或其它錯誤。同時,一旦對隊列的修改完成,必須立即釋放互斥鎖,以便其它線程可以訪問隊列。BHG28資訊網——每日最新資訊28at.com

然后是 not_empty 條件變量,當一個新元素被添加到隊列時,應該向 not_empty發送一個信號。這個動作會通知那些想從隊列中獲取元素,但因隊列為空而陷入阻塞的線程,現在隊列中已經有了新的元素,它們可以繼續執行獲取元素的操作。BHG28資訊網——每日最新資訊28at.com

接下來是 not_full 條件變量,當從隊列中取走一個元素時,應該向 not_full 發送一個信號。這個動作通知那些想往隊列添加元素,但因隊列已滿而陷入阻塞的線程,現在隊列中已經有了可用空間,它們可以繼續執行添加元素的操作。BHG28資訊網——每日最新資訊28at.com

最后是 all_tasks_done 條件變量,當處理的任務全部完成,即計數器 unfinished_task 為 0 時,應該向 all_tasks_done 發送一個信號。這個動作會通知那些執行了 join() 方法而陷入阻塞的線程,它們可以繼續往下執行了。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

因為線程隊列采用了雙端隊列存儲元素,所以雙端隊列的長度就是線程隊列的元素個數。如果元素個數為 0,那么隊列就是空;如果容量大于 0,并且小于等于元素個數,那么隊列就滿了。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

前面說了,put_nowait 和 get_nowait 本質上就是調用了 put 和 get,所以我們的重點是 put 和 get 兩個方法。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

以上就是 put 方法的底層實現,不難理解。說完了 put,再來看看 get。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

最后是 task_done 和 join 方法,看看它們的內部邏輯。BHG28資訊網——每日最新資訊28at.com

圖片圖片BHG28資訊網——每日最新資訊28at.com

調用 join 方法,當 unfinished_task 大于 0 時,會陷入阻塞。調用 task_done 方法,會將未完成任務數減 1,如果為 0,那么喚醒阻塞等待的線程。BHG28資訊網——每日最新資訊28at.com

需要注意的是,喚醒調用的方法不是 notify,而是 notify_all。對于添加元素和獲取元素,每次顯然只能喚醒一個線程,此時調用 notify。而 unfinished_task 為 0 時,應該要喚醒所有等待的線程,因此要調用 notify_all。BHG28資訊網——每日最新資訊28at.com

最后線程隊列也有相應的 PriorityQueue 和 LifoQueue,它們的用法、實現和協程里面的這兩個隊列是一樣的。BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

小結

BHG28資訊網——每日最新資訊28at.com

BHG28資訊網——每日最新資訊28at.com

以上便是協程隊列和線程隊列的具體用法和實現原理,它們本質上都是基于雙端隊列實現具體的元素存儲,并且在隊列已滿和隊列為空時,可以阻塞等待。BHG28資訊網——每日最新資訊28at.com

只不過協程隊列是通過 Future 對象實現的,而線程隊列是通過條件變量實現的。BHG28資訊網——每日最新資訊28at.com

當然,除了協程隊列和線程隊列,還有進程隊列,但進程隊列要復雜的多。因此關于進程隊列的實現細節,我們以后專門花篇幅去介紹。BHG28資訊網——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-38110-0.html源碼解密協程隊列和線程隊列的實現原理

聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com

上一篇: 被人說 Lambda 代碼像...,那是沒用下面這三個方法

下一篇: 一次 K8s 升級,竟然導致滴滴故障 12 小時?

標簽:
  • 熱門焦點
  • 紅魔電競平板評測:大屏幕硬實力

    前言:三年的疫情因為要上網課的原因激活了平板市場,如今網課的時代已經過去,大家的生活都恢復到了正軌,這也就意味著,真正考驗平板電腦生存的環境來了。也就是面對著這種殘酷的
  • iPhone賣不動了!蘋果股價創年內最大日跌幅:市值一夜蒸發萬億元

    8月5日消息,今天凌晨美股三大指數高開低走集體收跌,道指跌0.41%;納指跌0.36%;標普500指數跌0.52%。熱門科技股也都變化極大,其中蘋果報181.99美元,跌4.8%,創
  • 摸魚心法第一章——和配置文件說拜拜

    為了能摸魚我們團隊做了容器化,但是帶來的問題是服務配置文件很麻煩,然后大家在群里進行了“親切友好”的溝通圖片圖片圖片圖片對比就對比,簡單對比下獨立配置中心和k8s作為配
  • JavaScript學習 -AES加密算法

    引言在當今數字化時代,前端應用程序扮演著重要角色,用戶的敏感數據經常在前端進行加密和解密操作。然而,這樣的操作在網絡傳輸和存儲中可能會受到惡意攻擊的威脅。為了確保數據
  • 阿里瓴羊One推出背后,零售企業迎數字化新解

    作者:劉曠近年來隨著數字經濟的高速發展,各式各樣的SaaS應用服務更是層出不窮,但本質上SaaS大多局限于單一業務流層面,對用戶核心關切的增長問題等則沒有提供更好的解法。在Saa
  • 回歸OPPO兩年,一加贏了銷量,輸了品牌

    成為OPPO旗下主打性能的先鋒品牌后,一加屢創佳績。今年618期間,一加手機全渠道銷量同比增長362%,憑借一加 11、一加 Ace 2、一加 Ace 2V三款爆品,一加
  • OPPO K11評測:旗艦級IMX890加持 2000元檔最強影像手機

    【Techweb評測】中端機型用戶群體巨大,占了中國目前手機市場的大頭,一直以來都是各手機品牌的“必爭之地”,其中OPPO K系列機型一直以來都以高品質、
  • 英特爾Xe-HP項目終止,將專注Xe-HPC/HPG系列顯卡

    據10 月 31 日消息報道,英特爾高級副總裁兼加速計算系統和圖形事業部總經理 表示,Xe-HP“ Arctic Sound” 系列服務器 GPU 已經應用于 oneAPI devcloud 云服
  • 北京:科技教育體驗基地開始登記

      北京“科技館之城”科技教育體驗基地登記和認證工作日前啟動。首批北京科技教育體驗基地擬于2023年全國科普日期間掛牌,后續還將開展常態化登記?! ”本┛萍冀逃w驗基
Top 主站蜘蛛池模板: 湖北省| 新昌县| 美姑县| 临泽县| 波密县| 仁布县| 康定县| 江山市| 三江| 平武县| 永靖县| 舟山市| 岗巴县| 莫力| 梓潼县| 隆德县| 宝应县| 苍梧县| 甘洛县| 松潘县| 美姑县| 沾化县| 邛崃市| 成都市| 修文县| 延长县| 保亭| 佛教| 久治县| 平远县| 上犹县| 辛集市| 沙湾县| 漯河市| 广德县| 大荔县| 依兰县| 荔波县| 呼和浩特市| 普宁市| 邮箱|