多任務(wù)的實(shí)現(xiàn)有3種方式：

• 多進(jìn)程模式毡熏；
• 多線程模式；
• 多進(jìn)程+多線程模式酵镜。

多進(jìn)程

Unix/Linux操作系統(tǒng)提供了一個(gè) fork() 系統(tǒng)調(diào)用。調(diào)用一次柴钻，返回兩次淮韭，因?yàn)椴僮飨到y(tǒng)自動(dòng)把當(dāng)前進(jìn)程（稱為父進(jìn)程）復(fù)制了一份（稱為子進(jìn)程），然后顿颅，分別在父進(jìn)程和子進(jìn)程內(nèi)返回

子進(jìn)程永遠(yuǎn)返回0缸濒，而父進(jìn)程返回子進(jìn)程的ID

Python 的 os 模塊封裝了常見(jiàn)的系統(tǒng)調(diào)用，其中就包括

• os.fork() 創(chuàng)建子進(jìn)程
• os.getpid() 獲取自身 ID
• os.getppid() 獲取父進(jìn)程 ID

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))

運(yùn)行結(jié)果如下：

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

multiprocessing

multiprocessing 模塊就是跨平臺(tái)版本的多進(jìn)程模塊

multiprocessing 模塊提供了一個(gè) Process 類來(lái)代表一個(gè)進(jìn)程對(duì)象

from multiprocessing import Process
import os

# 子進(jìn)程要執(zhí)行的代碼
def run_proc(name):
    print('Run child process %s (%s)...' % (name, os.getpid()))

if __name__=='__main__':
    print('Parent process %s.' % os.getpid())
    
    # 新建一個(gè)子進(jìn)程
    p = Process(target=run_proc, args=('test',))
    print('Child process will start.')
    
    # 調(diào)用 start() 方法啟動(dòng)進(jìn)程
    p.start()
    # 阻塞當(dāng)前進(jìn)程粱腻，直到 p 進(jìn)程執(zhí)行完庇配，再繼續(xù)執(zhí)行當(dāng)前進(jìn)程，通常用于進(jìn)程間的同步
    p.join()
    print('Child process end.')

Pool

如果要啟動(dòng)大量的子進(jìn)程绍些，可以用進(jìn)程池的方式批量創(chuàng)建子進(jìn)程：

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())
    
    # 創(chuàng)建一個(gè)容量為4的進(jìn)程池
    # Pool的默認(rèn)大小是CPU的核數(shù)
    p = Pool(4)
    for i in range(5):
    
        # 放入子進(jìn)程要處理的函數(shù)
        p.apply_async(long_time_task, args=(i,))
    print('Waiting for all subprocesses done...')
    
    # 調(diào)用join()之前必須先調(diào)用close()捞慌，調(diào)用close()之后就不能繼續(xù)添加新的Process了
    p.close()
    
    # 對(duì)Pool對(duì)象調(diào)用join()方法會(huì)等待所有子進(jìn)程執(zhí)行完畢
    p.join()
    print('All subprocesses done.')

執(zhí)行結(jié)果如下：

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.

task 0，1柬批，2啸澡，3是立刻執(zhí)行的，而task 4要等待前面某個(gè)task完成后才執(zhí)行氮帐，最多同時(shí)執(zhí)行4個(gè)進(jìn)程

子進(jìn)程

很多時(shí)候嗅虏，子進(jìn)程并不是自身，而是一個(gè)外部進(jìn)程

我們創(chuàng)建了子進(jìn)程后上沐，還需要控制子進(jìn)程的輸入和輸出

subprocess 模塊可以讓我們非常方便地啟動(dòng)一個(gè)子進(jìn)程皮服，然后控制其輸入和輸出

下面的例子演示了如何在Python代碼中運(yùn)行命令nslookup www.python.org，這和 cmd 直接運(yùn)行的效果是一樣的：

import subprocess

print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)

如果子進(jìn)程還需要輸入，則可以通過(guò)communicate()方法輸入：

import subprocess

print('$ nslookup')

p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

output, err = p.communicate(b'set q=mx\npython.org\nexit\n')

print(output.decode('utf-8'))
print('Exit code:', p.returncode)

相當(dāng)于在 cmd 執(zhí)行命令 nslookup龄广，然后手動(dòng)輸入：

set q=mx
python.org
exit

進(jìn)程間通信

Python 的 multiprocessing 模塊包裝了底層的機(jī)制硫眯，提供了 Queue 、Pipes 等多種方式來(lái)交換數(shù)據(jù)

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

# 寫數(shù)據(jù)進(jìn)程執(zhí)行的代碼:
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())

# 讀數(shù)據(jù)進(jìn)程執(zhí)行的代碼:
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__':
    # 父進(jìn)程創(chuàng)建Queue择同，并傳給各個(gè)子進(jìn)程：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    
    # 啟動(dòng)子進(jìn)程pw两入，寫入:
    pw.start()
    
    # 啟動(dòng)子進(jìn)程pr，讀取:
    pr.start()
    
    # 等待pw結(jié)束:
    pw.join()
    
    # pr進(jìn)程里是死循環(huán)敲才，無(wú)法等待其結(jié)束裹纳，只能強(qiáng)行終止:
    pr.terminate()

多線程

Python 的線程是真正的 Posix Thread ，而不是模擬出來(lái)的線程

Python的標(biāo)準(zhǔn)庫(kù)提供了兩個(gè)模塊：_thread 和 threading 归斤，_thread 是低級(jí)模塊痊夭，threading 是高級(jí)模塊，對(duì) _thread 進(jìn)行了封裝

絕大多數(shù)情況下脏里，我們只需要使用 threading 這個(gè)高級(jí)模塊

任何進(jìn)程默認(rèn)就會(huì)啟動(dòng)一個(gè)線程，我們把該線程稱為主線程虹曙，主線程又可以啟動(dòng)新的線程

Python 的 threading 模塊有個(gè) current_thread() 函數(shù)迫横，它永遠(yuǎn)返回當(dāng)前線程的實(shí)例。主線程實(shí)例的名字叫 MainThread 酝碳，子線程的名字在創(chuàng)建時(shí)指定

import time, threading

# 新線程執(zhí)行的代碼:
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)

# 創(chuàng)建 Thread 實(shí)例
t = threading.Thread(target=loop, name='LoopThread')

t.start()
t.join()

print('thread %s ended.' % threading.current_thread().name)

Lock

多線程和多進(jìn)程最大的不同在于

• 多進(jìn)程中矾踱，同一個(gè)變量，各自有一份拷貝存在于每個(gè)進(jìn)程中疏哗，互不影響
• 多線程中呛讲，所有變量都由所有線程共享，所以返奉，任何一個(gè)變量都可以被任何一個(gè)線程修改

線程之間共享數(shù)據(jù)最大的危險(xiǎn)在于多個(gè)線程同時(shí)改一個(gè)變量贝搁，把內(nèi)容給改亂了

import time, threading

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

def change_it(n):
    # 先存后取，結(jié)果應(yīng)該為0:
    global balance
    # 1. 計(jì)算balance + n芽偏，存入臨時(shí)變量中雷逆；2. 將臨時(shí)變量的值賦給balance。
    balance = balance + n
    balance = balance - n

def run_thread(n):
    for i in range(100000):
        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)

當(dāng)代碼正常執(zhí)行時(shí)：

初始值 balance = 0

t1: x1 = balance + 5 # x1 = 0 + 5 = 5
t1: balance = x1     # balance = 5
t1: x1 = balance - 5 # x1 = 5 - 5 = 0
t1: balance = x1     # balance = 0

t2: x2 = balance + 8 # x2 = 0 + 8 = 8
t2: balance = x2     # balance = 8
t2: x2 = balance - 8 # x2 = 8 - 8 = 0
t2: balance = x2     # balance = 0

結(jié)果 balance = 0

t1和t2交替運(yùn)行時(shí)：

初始值 balance = 0

t1: x1 = balance + 5  # x1 = 0 + 5 = 5

t2: x2 = balance + 8  # x2 = 0 + 8 = 8
t2: balance = x2      # balance = 8

t1: balance = x1      # balance = 5
t1: x1 = balance - 5  # x1 = 5 - 5 = 0
t1: balance = x1      # balance = 0

t2: x2 = balance - 8  # x2 = 0 - 8 = -8
t2: balance = x2   # balance = -8

結(jié)果 balance = -8

所以污尉，我們必須確保一個(gè)線程在修改balance（即公共變量）的時(shí)候膀哲，別的線程一定不能改

創(chuàng)建一個(gè)鎖就是通過(guò) threading.Lock() 來(lái)實(shí)現(xiàn)：

import time, threading

balance = 0
lock = threading.Lock()

def change_it(n):
    # 先存后取，結(jié)果應(yīng)該為0:
    global balance
    # 1. 計(jì)算balance + n被碗，存入臨時(shí)變量中某宪；2. 將臨時(shí)變量的值賦給balance。
    balance = balance + n
    balance = balance - n

def run_thread(n):
    for i in range(100000):
        # 先要獲取鎖:
        lock.acquire()
        # 用try...finally來(lái)確保鎖一定會(huì)被釋放
        try:
            # 放心地改吧:
            change_it(n)
        finally:
            # 改完了一定要釋放鎖:
            lock.release()
            
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)

鎖的好處就是確保了某段關(guān)鍵代碼只能由一個(gè)線程從頭到尾完整地執(zhí)行

鎖的壞處首先是阻止了多線程并發(fā)執(zhí)行锐朴，包含鎖的某段代碼實(shí)際上只能以單線程模式執(zhí)行兴喂，效率就大大地下降了

其次，由于可以存在多個(gè)鎖，不同的線程持有不同的鎖瞻想，并試圖獲取對(duì)方持有的鎖時(shí)压真，可能會(huì)造成死鎖，導(dǎo)致多個(gè)線程全部掛起蘑险，既不能執(zhí)行滴肿，也無(wú)法結(jié)束，只能靠操作系統(tǒng)強(qiáng)制終止

多核CPU

一個(gè)死循環(huán)線程會(huì)100%占用一個(gè)CPU佃迄，但啟動(dòng)N個(gè)死循環(huán)線程泼差，卻不會(huì)使把N核CPU的核心全部跑滿

import multiprocessing

# 獲取 CPU 核數(shù)
multiprocessing.cpu_count()

因?yàn)镻ython的線程雖然是真正的線程，但解釋器執(zhí)行代碼時(shí)呵俏，有一個(gè)GIL鎖：Global Interpreter Lock堆缘，任何Python線程執(zhí)行前，必須先獲得GIL鎖普碎，然后吼肥，每執(zhí)行100條字節(jié)碼，解釋器就自動(dòng)釋放GIL鎖麻车，讓別的線程有機(jī)會(huì)執(zhí)行缀皱。

這個(gè)GIL全局鎖實(shí)際上把所有線程的執(zhí)行代碼都給上了鎖，所以动猬，多線程在Python中只能交替執(zhí)行啤斗，即使100個(gè)線程跑在100核CPU上，也只能用到1個(gè)核

Python雖然不能利用多線程實(shí)現(xiàn)多核任務(wù)赁咙，但可以通過(guò)多進(jìn)程實(shí)現(xiàn)多核任務(wù)钮莲。多個(gè)Python進(jìn)程有各自獨(dú)立的GIL鎖，互不影響

ThreadLocal

在多線程環(huán)境下彼水，使用自己的局部變量比使用全局變量好崔拥，但是局部變量也有問(wèn)題，就是在函數(shù)調(diào)用的時(shí)候猿涨，傳遞起來(lái)很麻煩

可以用全局 dict 存放所有的待傳遞對(duì)象握童，然后以 thread 自身作為 key 獲得線程對(duì)應(yīng)傳遞對(duì)象

ThreadLocal 對(duì)象不用查找 dict ，ThreadLocal 幫你自動(dòng)做這件事：

import threading

# 創(chuàng)建全局ThreadLocal對(duì)象:
local_school = threading.local()

def process_student():
    # 獲取當(dāng)前線程關(guān)聯(lián)的student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))
    
def process_thread(name):
    # 綁定ThreadLocal的student:
    local_school.student = name
    process_student()

t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

全局變量 local_school 就是一個(gè) ThreadLocal 對(duì)象叛赚，每個(gè) Thread 對(duì)它都可以讀寫 student 屬性澡绩，但互不影響

可以把 local_school 看成全局變量，但每個(gè)屬性如 local_school.student 都是線程的局部變量俺附，可以任意讀寫而互不干擾肥卡，也不用管理鎖的問(wèn)題，ThreadLocal 內(nèi)部會(huì)處理

ThreadLocal 最常用的地方就是為每個(gè)線程綁定一個(gè)數(shù)據(jù)庫(kù)連接事镣，HTTP請(qǐng)求步鉴，用戶身份信息等，這樣一個(gè)線程的所有調(diào)用到的處理函數(shù)都可以非常方便地訪問(wèn)這些資源

進(jìn)程 vs 線程

要實(shí)現(xiàn)多任務(wù)，通常我們會(huì)設(shè)計(jì)Master-Worker模式氛琢，Master負(fù)責(zé)分配任務(wù)喊递，Worker負(fù)責(zé)執(zhí)行任務(wù)

多進(jìn)程模式最大的優(yōu)點(diǎn)就是穩(wěn)定性高，因?yàn)橐粋€(gè)子進(jìn)程崩潰了阳似，不會(huì)影響主進(jìn)程和其他子進(jìn)程

多進(jìn)程模式的缺點(diǎn)是創(chuàng)建進(jìn)程的代價(jià)大骚勘，在Unix/Linux系統(tǒng)下，用 fork 調(diào)用還行撮奏，在Windows下創(chuàng)建進(jìn)程開(kāi)銷巨大俏讹。另外，操作系統(tǒng)能同時(shí)運(yùn)行的進(jìn)程數(shù)也是有限的畜吊，在內(nèi)存和CPU的限制下泽疆，如果有幾千個(gè)進(jìn)程同時(shí)運(yùn)行，操作系統(tǒng)連調(diào)度都會(huì)成問(wèn)題

多線程模式通常比多進(jìn)程快一點(diǎn)玲献，但是也快不到哪去

多線程模式致命的缺點(diǎn)就是任何一個(gè)線程掛掉都可能直接造成整個(gè)進(jìn)程崩潰殉疼，因?yàn)樗芯€程共享進(jìn)程的內(nèi)存

在Windows下，多線程的效率比多進(jìn)程要高

線程切換

無(wú)論是多進(jìn)程還是多線程青自，只要數(shù)量一多株依，效率肯定上不去

計(jì)算密集型 vs. IO密集型

是否采用多任務(wù)的第二個(gè)考慮是任務(wù)的類型。我們可以把任務(wù)分為計(jì)算密集型和IO密集型

計(jì)算密集型任務(wù)的特點(diǎn)是要進(jìn)行大量的計(jì)算延窜，消耗CPU資源。要最高效地利用CPU抹锄，計(jì)算密集型任務(wù)同時(shí)進(jìn)行的數(shù)量應(yīng)當(dāng)?shù)扔贑PU的核心數(shù)逆瑞。對(duì)于計(jì)算密集型任務(wù)，最好用C語(yǔ)言編寫伙单。

IO密集型任務(wù)的特點(diǎn)是CPU消耗很少获高，任務(wù)的大部分時(shí)間都在等待IO操作完成（因?yàn)镮O的速度遠(yuǎn)遠(yuǎn)低于CPU和內(nèi)存的速度）吵冒。對(duì)于IO密集型任務(wù)宣虾，任務(wù)越多稽寒，CPU效率越高结序，但也有一個(gè)限度吴裤。常見(jiàn)的大部分任務(wù)都是IO密集型任務(wù)燕锥，比如Web應(yīng)用顽频。

異步IO

如果充分利用操作系統(tǒng)提供的異步IO支持菱蔬，就可以用單進(jìn)程單線程模型來(lái)執(zhí)行多任務(wù)游两，這種全新的模型稱為事件驅(qū)動(dòng)模型砾层，Nginx就是支持異步IO的Web服務(wù)器

對(duì)應(yīng)到Python語(yǔ)言，單線程的異步編程模型稱為協(xié)程

分布式進(jìn)程

在Thread和Process中贱案，應(yīng)當(dāng)優(yōu)選Process肛炮，因?yàn)镻rocess更穩(wěn)定，而且，Process可以分布到多臺(tái)機(jī)器上侨糟，而Thread最多只能分布到同一臺(tái)機(jī)器的多個(gè)CPU上

Python的 multiprocessing 模塊不但支持多進(jìn)程碍扔，其中 managers 子模塊還支持把多進(jìn)程分布到多臺(tái)機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者秕重，將任務(wù)分布到其他多個(gè)進(jìn)程中不同，依靠網(wǎng)絡(luò)通信

服務(wù)進(jìn)程 master 負(fù)責(zé)啟動(dòng) Queue ，把 Queue 注冊(cè)到網(wǎng)絡(luò)上悲幅，然后往 Queue 里面寫入任務(wù)：

import random, time, queue
from multiprocessing.managers import BaseManager

# 發(fā)送任務(wù)的隊(duì)列:
task_queue = queue.Queue()
# 接收結(jié)果的隊(duì)列:
result_queue = queue.Queue()

# 從BaseManager繼承的QueueManager:
class QueueManager(BaseManager):
    pass
    
# 把兩個(gè)Queue都注冊(cè)到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)

# 綁定端口5000, 設(shè)置驗(yàn)證碼'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')

# 啟動(dòng)Queue:
manager.start()

# 獲得通過(guò)網(wǎng)絡(luò)訪問(wèn)的Queue對(duì)象:
task = manager.get_task_queue()
result = manager.get_result_queue()

# 放幾個(gè)任務(wù)進(jìn)去:
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)

# 從result隊(duì)列讀取結(jié)果:
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)

# 關(guān)閉:
manager.shutdown()
print('master exit.')

在分布式多進(jìn)程環(huán)境下套鹅，添加任務(wù)到Queue不可以直接對(duì)原始的 task_queue 進(jìn)行操作，那樣就繞過(guò)了 QueueManager 的封裝汰具，必須通過(guò) manager.get_task_queue() 獲得的 Queue 接口添加

在另一臺(tái)機(jī)器上啟動(dòng)任務(wù)進(jìn)程 worker（本機(jī)上啟動(dòng)也可以）：

import time, sys, queue
from multiprocessing.managers import BaseManager

# 創(chuàng)建類似的QueueManager:
class QueueManager(BaseManager):
    pass
    
# 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue卓鹿，所以注冊(cè)時(shí)只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 連接到服務(wù)器，也就是運(yùn)行task_master.py的機(jī)器:
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
# 端口和驗(yàn)證碼注意保持與task_master.py設(shè)置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 從網(wǎng)絡(luò)連接:
m.connect()

# 獲取Queue的對(duì)象:
task = m.get_task_queue()
result = m.get_result_queue()

# 從task隊(duì)列取任務(wù),并把結(jié)果寫入result隊(duì)列:
for i in range(10):
    try:
        n = task.get(timeout=1)
        print('run task %d * %d...' % (n, n))
        r = '%d * %d = %d' % (n, n, n*n)
        time.sleep(1)
        result.put(r)
    except Queue.Empty:
        print('task queue is empty.')
        
# 處理結(jié)束:
print('worker exit.')

Queue 之所以能通過(guò)網(wǎng)絡(luò)訪問(wèn)留荔，就是通過(guò) QueueManager 實(shí)現(xiàn)的吟孙。由于 QueueManager 管理的不止一個(gè) Queue ，所以聚蝶，要給每個(gè) Queue 的網(wǎng)絡(luò)調(diào)用接口起個(gè)名字杰妓，比如 get_task_queue

authkey 是為了保證兩臺(tái)機(jī)器正常通信，不被其他機(jī)器惡意干擾碘勉。如果 worker 的authkey和 master 的authkey不一致巷挥，肯定連接不上