python隊列與線程

A. 一文帶你讀懂python線程

Python線程

進程有很多優點，它提供了多道編程，可以提高計算機CPU的利用率。既然進程這么優秀，為什麼還要線程呢？其實，仔細觀察就會發現進程還是有很多缺陷的。

主要體現在一下幾個方面：

進程只能在一個時間做一個任務，如果想同時做兩個任務或多個任務，就必須開啟多個進程去完成多個任務。

進程在執行的過程中如果阻塞，例如等待輸入，整個進程就會掛起，即使進程中有些工作不依賴於輸入的數據，也將無法執行。

每個進程都有自己的獨立空間，所以多進程的創建，銷毀相比於多線程更加耗時，也更加佔用系統資源。

進程是資源分配的最小單位，線程是CPU調度的最小單位，每一個進程中至少有一個線程。

線程與進程的區別

可以歸納為以下4點：

1）地址空間：進程間相互獨立的每個進程都有自己獨立的內存空間，也就是說一個進程內的數據在另一個進程是不可見的。但同一進程中的各線程間數據是共享的。

2）通信：由於每個進程有自己獨立的內存空間，所以進程間通信需要IPC，而進程內的數據對於多個線程來說是共享的，每個線程都可以訪問，所以為了保證數據的一致性，需要使用鎖。

3）調度和切換：線程上下文切換比進程上下文切換要快得多。

4）在多線程操作系統中，進程不是一個可執行的實體，它主要的功能是向操作系統申請一塊內存空間，然後在內存空間中開線程來執行任務，相當於一個容器，容器中的線程才是真正的執行體。一個進程可以包含多個線程，而一個線程是不能包含進程的。因為進程是系統分配資源的最小單位，所以線程不能向操作系統申請自己的空間，但一個線程內可以包含多個線程。

相關推薦：《Python視頻教程》

線程的特點：

在多線程的操作系統中，通常是在一個進程中包括多個線程，每個線程都是作為利用CPU的基本單位，是花費最小開銷的實體。線程具有以下屬性。

1）輕型實體

線程中的實體基本上不擁有系統資源，只是有一點必不可少的、能保證獨立運行的資源。

線程的實體包括程序、數據和TCB。線程是動態概念，它的動態特性由線程式控制制塊TCB（Thread Control Block）描述。

2）獨立調度和分派的基本單位。

在多線程OS中，線程是能獨立運行的基本單位，因而也是獨立調度和分派的基本單位。由於線程很「輕」，故線程的切換非常迅速且開銷小（在同一進程中的）。

3）共享進程資源。

在同一進程中的各個線程，都可以共享該進程所擁有的資源，這首先表現在：所有線程都具有相同的進程id，這意味著，線程可以訪問該進程的每一個內存資源；此外，還可以訪問進程所擁有的已打開文件、定時器、信號量機構等。由於同一個進程內的線程共享內存和文件，所以線程之間互相通信不必調用內核。

4）可並發執行

在一個進程中的多個線程之間，可以並發執行，甚至允許在一個進程中所有線程都能並發執行；同樣，不同進程中的線程也能並發執行，充分利用和發揮了處理機與外圍設備並行工作的能力。

線程的實現可以分為兩類：

用戶級線程(User-Level Thread)和內核級線程(Kernel-Level Thread)，後者又稱為內核支持的線程或輕量級進程。在多線程操作系統中，各個系統的實現方式並不相同，在有的系統中實現了用戶級線程，有的系統中實現了內核級線程。

用戶線程和內核線程的區別：

1、內核支持線程是OS內核可感知的，而用戶級線程是OS內核不可感知的。

2、用戶級線程的創建、撤消和調度不需要OS內核的支持，是在語言（如Java）這一級處理的；而內核支持線程的創建、撤消和調度都需OS內核提供支持，而且與進程的創建、撤消和調度大體是相同的。

3、用戶級線程執行系統調用指令時將導致其所屬進程被中斷，而內核支持線程執行系統調用指令時，只導致該線程被中斷。

4、在只有用戶級線程的系統內，CPU調度還是以進程為單位，處於運行狀態的進程中的多個線程，由用戶程序控制線程的輪換運行；在有內核支持線程的系統內，CPU調度則以線程為單位，由OS的線程調度程序負責線程的調度。

5、用戶級線程的程序實體是運行在用戶態下的程序，而內核支持線程的程序實體則是可以運行在任何狀態下的程序。

內核線程的優缺點：

優點：當有多個處理機時，一個進程的多個線程可以同時執行。

缺點：由內核進行調度。

用戶線程的優缺點：

優點：

線程的調度不需要內核直接參與，控制簡單。

可以在不支持線程的操作系統中實現。

創建和銷毀線程、線程切換代價等線程管理的代價比內核線程少得多。

允許每個進程定製自己的調度演算法，線程管理比較靈活。

線程能夠利用的表空間和堆棧空間比內核級線程多。

同一進程中只能同時有一個線程在運行，如果有一個線程使用了系統調用而阻塞，那麼整個進程都會被掛起。另外，頁面失效也會產生同樣的問題。

缺點：

資源調度按照進程進行，多個處理機下，同一個進程中的線程只能在同一個處理機下分時復用。

B. Python實現簡單多線程任務隊列

Python實現簡單多線程任務隊列
最近我在用梯度下降演算法繪制神經網路的數據時，遇到了一些演算法性能的問題。梯度下降演算法的代碼如下（偽代碼）：
defgradient_descent(): # the gradient descent code plotly.write(X, Y)
一般來說，當網路請求 plot.ly 繪圖時會阻塞等待返回，於是也會影響到其他的梯度下降函數的執行速度。
一種解決辦法是每調用一次 plotly.write 函數就開啟一個新的線程，但是這種方法感覺不是很好。 我不想用一個像 cerely（一種分布式任務隊列）一樣大而全的任務隊列框架，因為框架對於我的這點需求來說太重了，並且我的繪圖也並不需要 redis 來持久化數據。
那用什麼辦法解決呢？我在 python 中寫了一個很小的任務隊列，它可以在一個單獨的線程中調用 plotly.write函數。下面是程序代碼。
classTaskQueue(Queue.Queue):
首先我們繼承 Queue.Queue 類。從 Queue.Queue 類可以繼承 get 和 put 方法，以及隊列的行為。
def__init__(self, num_workers=1): Queue.Queue.__init__(self) self.num_workers=num_workers self.start_workers()
初始化的時候，我們可以不用考慮工作線程的數量。
defadd_task(self, task,*args,**kwargs): args=argsor() kwargs=kwargsor{} self.put((task, args, kwargs))
我們把 task, args, kwargs 以元組的形式存儲在隊列中。*args 可以傳遞數量不等的參數，**kwargs 可以傳遞命名參數。
defstart_workers(self): foriinrange(self.num_workers): t=Thread(target=self.worker) t.daemon=True t.start()
我們為每個 worker 創建一個線程，然後在後台刪除。
下面是 worker 函數的代碼：
defworker(self): whileTrue: tupl=self.get() item, args, kwargs=self.get() item(*args,**kwargs) self.task_done()
worker 函數獲取隊列頂端的任務，並根據輸入參數運行，除此之外，沒有其他的功能。下面是隊列的代碼：
我們可以通過下面的代碼測試：
defblokkah(*args,**kwargs): time.sleep(5) print「Blokkah mofo!」 q=TaskQueue(num_workers=5) foriteminrange(1): q.add_task(blokkah) q.join()# wait for all the tasks to finish. print「Alldone!」
Blokkah 是我們要做的任務名稱。隊列已經緩存在內存中，並且沒有執行很多任務。下面的步驟是把主隊列當做單獨的進程來運行，這樣主程序退出以及執行資料庫持久化時，隊列任務不會停止運行。但是這個例子很好地展示了如何從一個很簡單的小任務寫成像工作隊列這樣復雜的程序。
defgradient_descent(): # the gradient descent code queue.add_task(plotly.write, x=X, y=Y)
修改之後，我的梯度下降演算法工作效率似乎更高了。如果你很感興趣的話，可以參考下面的代碼。 classTaskQueue(Queue.Queue): def__init__(self, num_workers=1):Queue.Queue.__init__(self)self.num_workers=num_workersself.start_workers() defadd_task(self, task,*args,**kwargs):args=argsor()kwargs=kwargsor{}self.put((task, args, kwargs)) defstart_workers(self):foriinrange(self.num_workers):t=Thread(target=self.worker)t.daemon=Truet.start() defworker(self):whileTrue:tupl=self.get()item, args, kwargs=self.get()item(*args,**kwargs)self.task_done() deftests():defblokkah(*args,**kwargs):time.sleep(5)print"Blokkah mofo!" q=TaskQueue(num_workers=5) foriteminrange(10):q.add_task(blokkah) q.join()# block until all tasks are doneprint"All done!" if__name__=="__main__":tests()

C. python 多線程消息隊列 怎麼避免 cpu空轉

不會，python目前不適應運行在多核上的多線程。僅能使用單核，如果需要多核你可以考慮多進程方式。

D. python有1000個url請求 放到queue 再十個線程 這樣輸出的數據是亂的 必須加鎖么

題主的問題表述不清，尤其十個線程做了哪些工作沒說明，我從字面猜測，是把1000個url放在隊列里，然後十個線程從隊列中取出url，請求之後，調用輸出函數。
先說隊列的問題，通常的設計里，queue只在添加元素，摘取元素的時候內部加鎖，以保證隊列數據不出錯，至於從queue中取出數據後怎麼處理，要不要放臨界區，queue是不管的，也不應該管，python的Queue模塊提供了相應的Queue類。
再說輸出結果是亂的，又是表述不清，如果你指的是最終輸出到一個文件里，文件內容亂了，那給你4種解決辦法：
1. 線程里每個url在請求到內容後，在寫文件時設立臨界區，保證串列輸出，強烈不建議這種方式。
2. 對每個url的請求內容寫一個文件，最後合並文件，這種最靈活。
3. 對每個工作線程的輸出寫一個文件，最後合並文件，最有並行style
4. 對每個url的請求內容再次寫到一個隊列里，另一個線程讀取此隊列寫文件，最有設計模式style
如果你要保證輸出結果是與輸入url同序的，建議你採用上面第2種辦法，合並文件的時候保證與url隊列同序

E. 如何用python簡單的設計開發非同步任務調度隊列

首先，客戶端可以直接扔任務到一個web services的介面上 –》 web api接收到任務後，會根據客戶端的ip和時間戳做task_id,返回給客戶，緊接著在redis裡面標記這任務的狀態。 格式為 func,args,kwargs,timeout=xx,queue_level=xx,interval_time=xx
主服務端:
一個線程，會不停的掃描那個redis hash表，取出任務的interval_time後，進行取模，如果匹配成功，就會塞到 redis sorted set有續集和裡面。
主線程，會不停的看看sorted set裡面，有沒有比自己實現小的任務，有的話，執行並刪除。 這里的執行是用多進程，為毛用多進程，因為線程很多時候是不好控制強制幹掉的。 每個任務都會用multiprocessing的方式去執行，去調用的時候，會多傳進一個task_id，用來把相關的進度推送到redis裡面。 另外，fork進程後，我會得到一個pid，我會把pid和timeout的信息，存放到kill_hash裡面。 然後會不間斷的查看，在指定的timeout內，這pid還在不在，如果還是存在，沒有退出的話，說明他的任務不太正常，我們就可以在main()，裡面幹掉這些任務。
所謂的優先順序就是個 High + middle +Low 的三合一鏈條而已，我每次都會堅持從高到低取任務，如果你的High級別的任務不斷的話，那麼我會一直幹不了低級別的任務了。 代碼的體現是在redis sorted set這邊，設立三個有序集合，我的worker隊列會從high開始做……
那麼如果想幹掉一個任務是如何操作的，首先我需要在 kill_hash 裡面標記任務應該趕緊幹掉，在就是在task_hash裡面把那個task_id幹掉，好讓他不會被持續的加入待執行的隊列裡面。

F. Python多線程中隊列到底是個什麼概念

你首先需要創建一個線程池，其次將數據壓進線程池，然後讀取線程池。

import Queue
ThrQueue = Queue.Queue()
ThrQueue.put('數據')
ThrQueue.get()

當然以上步驟需要一個class來支撐，那說起來就多了！

G. Python主要內容學的是什麼

第一步：Python開發基礎

Python全棧開發與人工智慧之Python開發基礎知識學習內容包括：Python基礎語法、數據類型、字元編碼、文件操作、函數、裝飾器、迭代器、內置方法、常用模塊等。

第二步：Python高級編程和資料庫開發

Python全棧開發與人工智慧之Python高級編程和資料庫開發知識學習內容包括：面向對象開發、Socket網路編程、線程、進程、隊列、IO多路模型、Mysql資料庫開發等。

第三步：前端開發

Python全棧開發與人工智慧之前端開發知識學習內容包括：Html、CSS、JavaScript開發、Jquery&bootstrap開發、前端框架VUE開發等。

第十步：高並發語言GO開發

Python全棧開發與人工智慧之高並發語言GO開發學習內容包括：GO語言基礎、數據類型與文件IO操作、函數和面向對象、並發編程等。

H. Python多進程運行——Multiprocessing基礎教程2

上篇文章簡單介紹了multiprocessing模塊，本文將要介紹進程之間的數據共享和信息傳遞的概念。

在多進程處理中，所有新創建的進程都會有這兩個特點：獨立運行，有自己的內存空間。

我們來舉個例子展示一下：

這個程序的輸出結果是：

在上面的程序中我們嘗試在兩個地方列印全局列表result的內容：

我們再用一張圖來幫助理解記憶不同進程間的數據關系：

如果程序需要在不同的進程之間共享一些數據的話，該怎麼做呢？不用擔心，multiprocessing模塊提供了Array對象和Value對象，用來在進程之間共享數據。

所謂Array對象和Value對象分別是指從共享內存中分配的ctypes數組和對象。我們直接來看一個例子，展示如何用Array對象和Value對象在進程之間共享數據：

程序輸出的結果如下：

成功了！主程序和p1進程輸出了同樣的結果，說明程序中確實完成了不同進程間的數據共享。那麼我們來詳細看一下上面的程序做了什麼：

在主程序中我們首先創建了一個Array對象：

向這個對象輸入的第一個參數是數據類型：i表示整數，d代表浮點數。第二個參數是數組的大小，在這個例子中我們創建了包含4個元素的數組。

類似的，我們創建了一個Value對象：

我們只對Value對象輸入了一個參數，那就是數據類型，與上述的方法一致。當然，我們還可以對其指定一個初始值（比如10），就像這樣：

隨後，我們在創建進程對象時，將剛創建好的兩個對象：result和square_sum作為參數輸入給進程：

在函數中result元素通過索引進行數組賦值，square_sum通過 value 屬性進行賦值。

注意：為了完整列印result數組的結果，需要使用 result[:] 進行列印，而square_sum也需要使用 value 屬性進行列印：

每當python程序啟動時，同時也會啟動一個伺服器進程。隨後，只要我們需要生成一個新進程，父進程就會連接到伺服器並請求它派生一個新進程。這個伺服器進程可以保存Python對象，並允許其他進程使用代理來操作它們。

multiprocessing模塊提供了能夠控制伺服器進程的Manager類。所以，Manager類也提供了一種創建可以在不同流程之間共享的數據的方法。

伺服器進程管理器比使用共享內存對象更靈活，因為它們可以支持任意對象類型，如列表、字典、隊列、值、數組等。此外，單個管理器可以由網路上不同計算機上的進程共享。

但是，伺服器進程管理器的速度比使用共享內存要慢。

讓我們來看一個例子：

這個程序的輸出結果是：

我們來理解一下這個程序做了什麼：首先我們創建了一個manager對象

在with語句下的所有行，都是在manager對象的范圍內的。接下來我們使用這個manager對象創建了列表（類似的，我們還可以用 manager.dict() 創建字典）。

最後我們創建了進程p1（用於在records列表中插入一條新的record）和p2（將records列印出來），並將records作為參數進行傳遞。

伺服器進程的概念再次用下圖總結一下：

為了能使多個流程能夠正常工作，常常需要在它們之間進行一些通信，以便能夠劃分工作並匯總最後的結果。multiprocessing模塊支持進程之間的兩種通信通道：Queue和Pipe。

使用隊列來回處理多進程之間的通信是一種比較簡單的方法。任何Python對象都可以使用隊列進行傳遞。我們來看一個例子：

上面這個程序的輸出結果是：

我們來看一下上面這個程序到底做了什麼。首先我們創建了一個Queue對象：

然後，將這個空的Queue對象輸入square_list函數。該函數會將列表中的數平方，再使用 put() 方法放入隊列中：

隨後使用 get() 方法，將q列印出來，直至q重新稱為一個空的Queue對象：

我們還是用一張圖來幫助理解記憶：

一個Pipe對象只能有兩個端點。因此，當進程只需要雙向通信時，它會比Queue對象更好用。

multiprocessing模塊提供了 Pipe() 函數，該函數返回由管道連接的一對連接對象。 Pipe() 返回的兩個連接對象分別表示管道的兩端。每個連接對象都有 send() 和 recv() 方法。

我們來看一個例子：

上面這個程序的輸出結果是：

我們還是來看一下這個程序到底做了什麼。首先創建了一個Pipe對象：

與上文說的一樣，該對象返回了一對管道兩端的兩個連接對象。然後使用 send() 方法和 recv() 方法進行信息的傳遞。就這么簡單。在上面的程序中，我們從一端向另一端發送一串消息。在另一端，我們收到消息，並在收到END消息時退出。

要注意的是，如果兩個進程(或線程)同時嘗試從管道的同一端讀取或寫入管道中的數據，則管道中的數據可能會損壞。不過不同的進程同時使用管道的兩端是沒有問題的。還要注意，Queue對象在進程之間進行了適當的同步，但代價是增加了計算復雜度。因此，Queue對象對於線程和進程是相對安全的。

最後我們還是用一張圖來示意：

Python的multiprocessing模塊還剩最後一篇文章：多進程的同步與池化

敬請期待啦！

I. 為什麼在python里推薦使用多進程而不是多線程

監控一個信號就起一個線程與進程處理。這樣的邏輯是不太合適的。所有的資源都是有限的，如果這樣浪費很快會資源管理失控。

常規的做法是起一個線程池，或者是進程池。 使用線程還是進程取決於你處理的信號的類型。如果計算量大，則需要進程池，如果只是設備等待，比如網路數據收發，則線程也勉強夠用。

信號過來後處理方法有兩種，一種是實時處理，這個沒有好辦法，可以用「微線程」的辦法做，盡量減少處理周期。另外一種是允許少量的延遲。那麼通常的做法是用隊列。將信號放到線程或者是進程池的消息隊列里。然後再由後者分配。

還有一種高效的處理方法，根據信號的值做hash，然後自動分發到不同的CPU或者是伺服器。這個就算是大規模並發處理機制。

通常情況下，比如一個WEB伺服器，它需要獲取一個請求，然後處理響應，可以使用線程模型，或者是進程模型。也是使用典型的池的方法。一個Pool的大於，取決於你的計算 機的計算 能力，內存大小，以及你的並發訪問數量。

所要要啟用多少個呢？假設你的一個信號的處理周期是1秒，你同時有100個信號進來，那麼就需要100個線程或者是進程。