Python學習記錄-多進程和多執行緒
- 2020 年 1 月 8 日
- 筆記
Python學習記錄-多進程和多執行緒
[TOC]
1. 進程和執行緒
進程
狹義定義：進程是正在運行的程式的實例（an instance of a computer program that is being executed）。 廣義定義：進程是一個具有一定獨立功能的程式關於某個數據集合的一次運行活動。它是作業系統動態執行的基本單元，在傳統的作業系統中，進程既是基本的分配單元，也是基本的執行單元。
執行緒
執行緒是作業系統能夠進行運算調度的最小單位。它被包含在進程之中，是進程中的實際運作單位。一條執行緒指的是進程中一個單一順序的控制流，一個進程中可以並發多個執行緒，每條執行緒並行執行不同的任務。
執行緒與進程比較
執行緒與進程的區別： 1）地址空間和其它資源（如打開文件）：進程間相互獨立，同一進程的各執行緒間共享。某進程內的執行緒在其它進程不可見。 2）通訊：進程間通訊IPC，執行緒間可以直接讀寫進程數據段（如全局變數）來進行通訊——需要進程同步和互斥手段的輔助，以保證數據的一致性。 3）創建：創建新執行緒很簡單，創建新進程需要對父進程進行一次克隆。 4）調度和切換：一個執行緒可以控制和操作同一進程里的其它執行緒，但是進程只能操作子進程；執行緒上下文切換比進程上下文切換要快得多。 5）在多執行緒OS中，進程不是一個可執行的實體。
<font color="red">注意：</font> 執行緒和進程快慢無法對比，因為執行緒被包含在進程中。
2. threading模組
執行緒有2種調用方式，如下：
直接調用
import threading import time def sayhi(num): # 定義每個執行緒要運行的函數 print("running on number:%s" % num) time.sleep(3) if __name__ == '__main__': t1 = threading.Thread(target=sayhi, args=(1,)) # 生成一個執行緒實例 t2 = threading.Thread(target=sayhi, args=(2,)) # 生成另一個執行緒實例 t1.start() # 啟動執行緒 t2.start() # 啟動另一個執行緒 print(t1.getName()) # 獲取執行緒名 print(t2.getName())繼承式調用
import threading import time class MyThread(threading.Thread): def __init__(self, num): threading.Thread.__init__(self) self.num = num def run(self): # 定義每個執行緒要運行的函數 print("running on number:%s" % self.num) time.sleep(3) if __name__ == '__main__': t1 = MyThread(1) t2 = MyThread(2) t1.start() t2.start()平常用到的主要方法：
start 執行緒準備就緒，等待CPU調度 setName 為執行緒設置名稱 getName 獲取執行緒名稱 setDaemon 設置為後台執行緒或前台執行緒（默認） 如果是後台執行緒，主執行緒執行過程中，後台執行緒也在進行，主執行緒執行完畢後，後台執行緒不論成功與否，均停止 如果是前台執行緒，主執行緒執行過程中，前台執行緒也在進行，主執行緒執行完畢後，等待前台執行緒也執行完成後，程式停止 join 逐個執行每個執行緒，執行完畢後繼續往下執行，該方法使得多執行緒變得無意義 run 執行緒被cpu調度後自動執行執行緒對象的run方法
2.1 Join & Daemon
import time import threading def run(n): print('[%s]------running----n' % n) time.sleep(2) print('--done--') def main(): for i in range(5): t = threading.Thread(target=run, args=[i, ]) t.start() t.join(1) print('starting thread', t.getName()) m = threading.Thread(target=main, args=[]) m.setDaemon(True) # 將main執行緒設置為Daemon執行緒,它做為程式主執行緒的守護執行緒,當主執行緒退出時,m執行緒也會退出,由m啟動的其它子執行緒會同時退出,不管是否執行完任務 m.start() m.join(timeout=2) print("---main thread done----")import time import threading def addNum(): global num # 在每個執行緒中都獲取這個全局變數 print('--get num:', num) time.sleep(1) num -= 1 # 對此公共變數進行-1操作 num = 100 # 設定一個共享變數 thread_list = [] for i in range(100): t = threading.Thread(target=addNum) t.start() thread_list.append(t) for t in thread_list: # 等待所有執行緒執行完畢 t.join() print('final num:', num)2.2 執行緒鎖(互斥鎖Mutex)
一個進程下可以啟動多個執行緒，多個執行緒共享父進程的記憶體空間，也就意味著每個執行緒可以訪問同一份數據，由於執行緒之間是進行隨機調度，並且每個執行緒可能只執行n條執行之後，當多個執行緒同時修改同一條數據時可能會出現臟數據，所以，出現了執行緒鎖 - 同一時刻允許一個執行緒執行操作。
import time import threading def addNum(): global num # 在每個執行緒中都獲取這個全局變數 print('--get num:', num) time.sleep(1) num -= 1 # 對此公共變數進行-1操作 num = 100 # 設定一個共享變數 thread_list = [] for i in range(100): t = threading.Thread(target=addNum) t.start() thread_list.append(t) for t in thread_list: # 等待所有執行緒執行完畢 t.join() print('final num:', num)因為python2.7以上版本，已經自動添加執行緒鎖，所以我們不用細糾。
import time import threading def addNum(): global num # 在每個執行緒中都獲取這個全局變數 print('--get num:', num) time.sleep(1) lock.acquire() # 修改數據前加鎖 num -= 1 # 對此公共變數進行-1操作 lock.release() # 修改後釋放 num = 100 # 設定一個共享變數 thread_list = [] lock = threading.Lock() # 生成全局鎖 for i in range(100): t = threading.Thread(target=addNum) t.start() thread_list.append(t) for t in thread_list: # 等待所有執行緒執行完畢 t.join() print('final num:', num)2.3 訊號量(Semaphore)
互斥鎖 同時只允許一個執行緒更改數據，而Semaphore是同時允許一定數量的執行緒更改數據 ，比如廁所有3個坑，那最多只允許3個人上廁所，後面的人只能等裡面有人出來了才能再進去。
import threading, time def run(n): semaphore.acquire() time.sleep(1) print("run the thread: %s" % n) semaphore.release() if __name__ == '__main__': num = 0 semaphore = threading.BoundedSemaphore(5) # 最多允許5個執行緒同時運行 for i in range(20): t = threading.Thread(target=run, args=(i,)) t.start()2.4 事件(event)
python執行緒的事件用於主執行緒控制其他執行緒的執行，事件主要提供了三個方法 set、wait、clear。
事件處理的機制：全局定義了一個「Flag」，如果「Flag」值為 False，那麼當程式執行 event.wait 方法時就會阻塞，如果「Flag」值為True，那麼event.wait 方法時便不再阻塞。
clear：將「Flag」設置為False set：將「Flag」設置為True
import threading def do(event): print('start') event.wait() print('execute') event_obj = threading.Event() for i in range(10): t = threading.Thread(target=do, args=(event_obj,)) t.start() event_obj.clear() inp = input('input:') if inp == 'true': event_obj.set()2.5 條件(Condition)
使得執行緒等待，只有滿足某條件時，才釋放n個執行緒。
未使用條件時：
import threading def run(n): con.acquire() con.wait() print("run the thread: %s" %n) con.release() if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start() while True: inp = input('>>>') if inp == 'q': break con.acquire() con.notify(int(inp)) con.release()使用條件時：
def condition_func(): ret = False inp = input('>>>') if inp == '1': ret = True return ret def run(n): con.acquire() con.wait_for(condition_func) print("run the thread: %s" %n) con.release() if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start()2.6 定時器(Timer)
定時器，指定n秒後執行某操作
from threading import Timer def hello(): print("hello, world") t = Timer(1, hello) t.start() # after 1 seconds, "hello, world" will be printed3. queue隊列模組
queue特別用於在多個執行緒間安全的交換。
class queue.Queue(maxsize=0) #先入先出 class queue.LifoQueue(maxsize=0) #後入先出 class queue.PriorityQueue(maxsize=0) #存儲數據時可設置優先順序的隊列
maxsize為隊列最大容量，如果設置為0或者小於0，則隊列大小無限制。
3.1 一些常用方法
task_done() 意味著之前入隊的一個任務已經完成。由隊列的消費者執行緒調用。每一個get()調用得到一個任務，接下來的task_done()調用告訴隊列該任務已經處理完畢。
如果當前一個join()正在阻塞，它將在隊列中的所有任務都處理完時恢復執行（即每一個由put()調用入隊的任務都有一個對應的task_done()調用。
join() 阻塞調用執行緒，直到隊列中的所有任務被處理掉。
只要有數據被加入隊列，未完成的任務數就會增加。當消費者執行緒調用task_done()（意味著有消費者取得任務並完成任務），未完成的任務數就會減少。當未完成的任務數降到0，join()解除阻塞。
put(item[, block[, timeout]]) 將item放入隊列中。
如果可選的參數block為True且timeout為空對象（默認的情況，阻塞調用，無超時）。 如果timeout是個正整數，阻塞調用進程最多timeout秒，如果一直無空空間可用，拋出Full異常（帶超時的阻塞調用）。 如果block為False，如果有空閑空間可用將數據放入隊列，否則立即拋出Full異常 其非阻塞版本為put_nowait等同於put(item, False)
get([block[, timeout]]) 從隊列中移除並返回一個數據。block跟timeout參數同put方法
其非阻塞方法為get_nowait()相當與get(False)
empty() 如果隊列為空，返回True,反之返回False
4. 生產者消費者模型
在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡生產執行緒和消費執行緒的工作能力來提高程式的整體處理數據的速度。
為什麼要使用生產者和消費者模式？
在執行緒世界裡，生產者就是生產數據的執行緒，消費者就是消費數據的執行緒。在多執行緒開發當中，如果生產者處理速度很快，而消費者處理速度很慢，那麼生產者就必須等待消費者處理完，才能繼續生產數據。同樣的道理，如果消費者的處理能力大於生產者，那麼消費者就必須等待生產者。為了解決這個問題於是引入了生產者和消費者模式。
什麼是生產者消費者模式？
生產者消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產者生產完數據之後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列里取，阻塞隊列就相當於一個緩衝區，平衡了生產者和消費者的處理能力。
生產者消費者例子：
import threading import queue import time def producer(name): count = 1 while True: for i in range(10): q.put("包子 %s" % count) print("生產了包子", count) count += 1 time.sleep(0.5) def consumer(n): while True: if q.qsize() > 0: print("%s 取到" %n , q.get()) time.sleep(1.5) q = queue.Queue(maxsize=10) p = threading.Thread(target=producer,args=("包子店",)) p.start() c1 = threading.Thread(target=consumer,args=("Lily",)) c1.start() c2 = threading.Thread(target=consumer,args=("Aeny",)) c2.start() c3 = threading.Thread(target=consumer,args=("Bob",)) c3.start()4.1 多執行緒的使用場景
python多執行緒不適合CPU密集型的任務，適合IO密集型的任務。
5. 多進程
多進程模組multiprocessing的基本使用方法：
from multiprocessing import Process import threading import time def thread_run(): print(threading.get_ident()) def foo(i): time.sleep(1) print('say hi',i) t = threading.Thread(target=thread_run,) t.start() for i in range(10): p = Process(target=foo,args=(i,)) p.start() # p.join()<font color="red">注意</font>： 由於進程之間的數據需要各自持有一份，所以創建進程需要的非常大的開銷。
獲取進程ID和父進程ID的方法示例：
from multiprocessing import Process import os def info(title): print(title) print('module name:', __name__) print('parent process:', os.getppid()) print('process id:', os.getpid()) print("nn") def f(name): info('33[31;1mfunction f33[0m') print('hello', name) if __name__ == '__main__': info('33[32;1mmain process line33[0m') p = Process(target=f, args=('bob',)) p.start() p.join()5.1 進程間數據交互
不同進程間記憶體是不共享的，要想實現兩個進程間的數據交換，可以用以下方法： Queues
from multiprocessing import Process, Queue def f(q): q.put([100, None, 'hello']) if __name__ == '__main__': q = Queue() p = Process(target=f, args=(q,)) p.start() print(q.get()) # prints "[100, None, 'hello']" p.join()
Pipes
from multiprocessing import Process, Pipe def f(conn): conn.send([1, True, 'first']) conn.send([2, True, 'second']) print("from parent:", conn.recv()) # from parent: Third conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) # [1, True, 'first'] print(parent_conn.recv()) # [2, True, 'second'] parent_conn.send("Third") p.join()5.2 進程數據共享
進程各自持有一份數據，默認無法共享數據。
from multiprocessing import Process import time li = [] def foo(i): li.append(i) print('say hi',li) for i in range(10): p = Process(target=foo,args=(i,)) p.start() print ('ending',li)結果類似這樣（每次的結果可能排序會有變化）。
say hi [2] say hi [3] say hi [5] say hi [0] say hi [1] say hi [4] say hi [6] say hi [7] say hi [8] ending [] say hi [9]
想要進程間共享數據
方法一：使用Array
from multiprocessing import Process,Array temp = Array('i', [11,22,33,44]) def Foo(i): temp[i] = 100+i for item in temp: print(i,'----->',item) for i in range(2): p = Process(target=Foo,args=(i,)) p.start() 方法二：manage.dict()共享數據
from multiprocessing import Process,Manager import os def f(d, l): d[os.getpid()] = os.getpid() l.append(os.getpid()) print(l) if __name__ == '__main__': manager = Manager() d= manager.dict() l = manager.list(range(5)) p_list = [] for i in range(10): p = Process(target=f,args=(d, l)) p.start() p_list.append(p) for res in p_list: # 等待結果 res.join() print(d) print(l)
5.3 進程同步
進程鎖的作用是保證螢幕列印正常，因為不加鎖有可能同一時間列印的數據導致顯示錯亂。
from multiprocessing import Process, Lock def f(l, i): l.acquire() try: print('hello world', i) finally: l.release() if __name__ == '__main__': lock = Lock() for num in range(10): Process(target=f, args=(lock, num)).start()5.4 進程池
進程池內部維護一個進程式列，當使用時，則去進程池中獲取一個進程，如果進程池序列中沒有可供使用的進進程，那麼程式就會等待，直到進程池中有可用進程為止。
進程池中有兩個方法：
- apply
- apply_async
from multiprocessing import Process, Pool import time import os def Foo(i): time.sleep(2) print("in process", os.getpid()) return i + 100 def Bar(arg): print('-->exec done:', arg, os.getpid()) if __name__ == '__main__': pool = Pool(processes=3) # 允許進程池同時放入的個數 print("主進程", os.getpid()) for i in range(10): pool.apply_async(func=Foo, args=(i,), callback=Bar) # callback=回調 # pool.apply(func=Foo, args=(i,)) # 串列 # pool.apply_async(func=Foo, args=(i,)) # 並行 print('end') pool.close() pool.join() # 進程池中進程執行完畢後再關閉，如果注釋，那麼進程直接關閉
