轉載於：https://www.liaoxuefeng.com/

程序

在LInux或Unix上作業系統提供了一個fork()函式，fork()函式呼叫一次返回兩次，因為作業系統自動把當前程序（稱為父程序）複製了一份（稱為子程序），然後，分別在父程序和子程序內返回。

例項

import os

print('Process (%s) start...' % os.getpid())
# Only works on Unix/Linux/Mac:
pid = os.fork()
if pid == 0:
    print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
    print('I (%s) just created a child process (%s).' % (os.getpid(), pid))

執行結果

Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.

multiprocessing

由於windows上沒有fork()函式呼叫，所以想要在windwos平臺上使用多程序就要藉助multiprocessing模組，multiprocessing模組就是跨平臺版本的多程序模組。

multiprocessing模組提供了一個Process類來代表一個程序物件

例項

from multiprocessing import Process
import os

# 子程序要執行的程式碼
def run_proc(name):
    print(
 
'Run child process %s (%s)...' % (name, os.getpid()))

if __name__=='__main__':
    print('Parent process %s.' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('Child process will start.')
    p.start()
    p.join()
    print('Child process end.')

執行結果

Parent process 928.
Child process will start.
Run child process test (
 
929)...
Process end.

其中join()方法是等待子程序結束之後再繼續往下執行，通常用於程序間同步。

Pool

如果要啟動大量的子程序，可以用程序池的方式批量建立子程序：

例項

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print('Run task %s (%s)...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('Task %s runs %0.2f seconds.' % (name, (end - start)))

if __name__=='__main__':
    print('Parent process %s.' % os.getpid())
    p = Pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print('Waiting for all subprocesses done...')
    p.close()
    p.join()
    print('All subprocesses done.')

執行結果

Parent process 669.
Waiting for all subprocesses done...
Run task 0 (671)...
Run task 1 (672)...
Run task 2 (673)...
Run task 3 (674)...
Task 2 runs 0.14 seconds.
Run task 4 (673)...
Task 1 runs 0.27 seconds.
Task 3 runs 0.86 seconds.
Task 0 runs 1.41 seconds.
Task 4 runs 1.91 seconds.
All subprocesses done.

程序間通訊

Process之間肯定是需要通訊的，作業系統提供了很多機制來實現程序間的通訊。Python的multiprocessing模組包裝了底層的機制，提供了Queue、Pipes等多種方式來交換資料。

我們以Queue為例，在父程序中建立兩個子程序，一個往Queue裡寫資料，一個從Queue裡讀資料：

例項

from multiprocessing import Process, Queue
import os, time, random

# 寫資料程序執行的程式碼:
def write(q):
    print('Process to write: %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print('Put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())

# 讀資料程序執行的程式碼:
def read(q):
    print('Process to read: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('Get %s from queue.' % value)

if __name__=='__main__':
    # 父程序建立Queue，並傳給各個子程序：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 啟動子程序pw，寫入:
    pw.start()
    # 啟動子程序pr，讀取:
    pr.start()
    # 等待pw結束:
    pw.join()
    # pr程序裡是死迴圈，無法等待其結束，只能強行終止:
    pr.terminate()

執行結果

Process to write: 50563
Put A to queue...
Process to read: 50564
Get A from queue.
Put B to queue...
Get B from queue.
Put C to queue...
Get C from queue.

執行緒

Python的標準庫提供了兩個模組：_thread和threading，_thread是低階模組，threading是高階模組，對_thread進行了封裝。絕大多數情況下，我們只需要使用threading這個高階模組。

啟動一個執行緒就是把一個函式傳入並建立Thread例項，然後呼叫start()開始執行：

例項

import time, threading

# 新執行緒執行的程式碼:
def loop():
    print('thread %s is running...' % threading.current_thread().name)
    n = 0
    while n < 5:
        n = n + 1
        print('thread %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('thread %s ended.' % threading.current_thread().name)

print('thread %s is running...' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('thread %s ended.' % threading.current_thread().name)

執行結果

thread MainThread is running...
thread LoopThread is running...
thread LoopThread >>> 1
thread LoopThread >>> 2
thread LoopThread >>> 3
thread LoopThread >>> 4
thread LoopThread >>> 5
thread LoopThread ended.
thread MainThread ended.

Lock

多執行緒和多程序最大的不同在於，多程序中，同一個變數，各自有一份拷貝存在於每個程序中，互不影響，而多執行緒中，所有變數都由所有執行緒共享，所以，任何一個變數都可以被任何一個執行緒修改，因此，執行緒之間共享資料最大的危險在於多個執行緒同時改一個變數，把內容給改亂了。

來看看多個執行緒同時操作一個變數怎麼把內容給改亂了：

例項

import time, threading

# 假定這是你的銀行存款:
balance = 0

def change_it(n):
    # 先存後取，結果應該為0:
    global balance
    balance = balance + n
    balance = balance - n

def run_thread(n):
    for i in range(2000000):
        change_it(n)

t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)

我們定義了一個共享變數balance，初始值為0，並且啟動兩個執行緒，先存後取，理論上結果應該為0，但是，由於執行緒的排程是由作業系統決定的，當t1、t2交替執行時，只要迴圈次數足夠多，balance的結果就不一定是0了。

究其原因，是因為修改balance需要多條語句，而執行這幾條語句時，執行緒可能中斷，從而導致多個執行緒把同一個物件的內容改亂了。

兩個執行緒同時一存一取，就可能導致餘額不對，你肯定不希望你的銀行存款莫名其妙地變成了負數，所以，我們必須確保一個執行緒在修改balance的時候，別的執行緒一定不能改。

如果我們要確保balance計算正確，就要給change_it()上一把鎖，當某個執行緒開始執行change_it()時，我們說，該執行緒因為獲得了鎖，因此其他執行緒不能同時執行change_it()，只能等待，直到鎖被釋放後，獲得該鎖以後才能改。由於鎖只有一個，無論多少執行緒，同一時刻最多隻有一個執行緒持有該鎖，所以，不會造成修改的衝突。建立一個鎖就是通過threading.Lock()來實現：

balance = 0
lock = threading.Lock()

def run_thread(n):
    for i in range(100000):
        # 先要獲取鎖:
        lock.acquire()
        try:
            # 放心地改吧:
            change_it(n)
        finally:
            # 改完了一定要釋放鎖:
            lock.release()

當多個執行緒同時執行lock.acquire()時，只有一個執行緒能成功地獲取鎖，然後繼續執行程式碼，其他執行緒就繼續等待直到獲得鎖為止。

獲得鎖的執行緒用完後一定要釋放鎖，否則那些苦苦等待鎖的執行緒將永遠等待下去，成為死執行緒。所以我們用try...finally來確保鎖一定會被釋放。

鎖的好處就是確保了某段關鍵程式碼只能由一個執行緒從頭到尾完整地執行，壞處當然也很多，首先是阻止了多執行緒併發執行，包含鎖的某段程式碼實際上只能以單執行緒模式執行，效率就大大地下降了。其次，由於可以存在多個鎖，不同的執行緒持有不同的鎖，並試圖獲取對方持有的鎖時，可能會造成死鎖，導致多個執行緒全部掛起，既不能執行，也無法結束，只能靠作業系統強制終止。

__EOF__