Python中編寫併發程序

GIL

在Python中,因爲歷史緣由(GIL),使得Python中多線程的效果很是不理想.GIL使得任什麼時候刻Python只能利用一個CPU核,而且它的調度算法簡單粗暴:多線程中,讓每一個線程運行一段時間t,而後強行掛起該線程,繼而去運行其餘線程,如此周而復始,直到全部線程結束.

這使得沒法有效利用計算機系統中的"局部性",頻繁的線程切換也對緩存不是很友好,形成資源的浪費.

聽說Python官方曾經實現了一個去除GIL的Python解釋器,可是其效果還不若有GIL的解釋器,遂放棄.後來Python官方推出了"利用多進程替代多線程"的方案,在Python3中也有concurrent.futures這樣的包,讓咱們的程序編寫能夠作到"簡單和性能兼得".

多進程/多線程+Queue

通常來講,在Python中編寫併發程序的經驗是:計算密集型任務使用多進程,IO密集型任務使用多進程或者多線程.另外,由於涉及到資源共享,因此須要同步鎖等一系列麻煩的步驟,代碼編寫不直觀.另一種好的思路是利用多進程/多線程+Queue的方法,能夠避免加鎖這樣麻煩低效的方式.

如今在Python2中利用Queue+多進程的方法來處理一個IO密集型任務.
假設如今須要下載多個網頁內容並進行解析,單進程的方式效率很低,因此使用多進程/多線程勢在必行.
咱們能夠先初始化一個tasks隊列,裏面將要存儲的是一系列dest_url,同時開啓4個進程向tasks中取任務而後執行,處理結果存儲在一個results隊列中,最後對results中的結果進行解析.最後關閉兩個隊列.

下面是一些主要的邏輯代碼.

python# -*- coding:utf-8 -*-

#IO密集型任務
#多個進程同時下載多個網頁
#利用Queue+多進程
#因爲是IO密集型,因此一樣能夠利用threading模塊

import multiprocessing

def main():
    tasks = multiprocessing.JoinableQueue()
    results = multiprocessing.Queue()
    cpu_count = multiprocessing.cpu_count()  #進程數目==CPU核數目

    create_process(tasks, results, cpu_count)   #主進程立刻建立一系列進程,可是因爲阻塞隊列tasks開始爲空,副進程所有被阻塞
    add_tasks(tasks)  #開始往tasks中添加任務
    parse(tasks, results)  #最後主進程等待其餘線程處理完成結果


def create_process(tasks, results, cpu_count):
    for _ in range(cpu_count):
        p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根據_worker建立對應的進程
        p.daemon = True  #讓全部進程能夠隨主進程結束而結束
        p.start() #啓動

def _worker(tasks, results):
    while True:   #由於前面全部線程都設置了daemon=True,故不會無限循環
        try:
            task = tasks.get()   #若是tasks中沒有任務,則阻塞
            result = _download(task)
            results.put(result)   #some exceptions do not handled
        finally:
            tasks.task_done()

def add_tasks(tasks):
    for url in get_urls():  #get_urls() return a urls_list
        tasks.put(url)

def parse(tasks, results):
    try: 
        tasks.join()
    except KeyboardInterrupt as err:
        print "Tasks has been stopped!"
        print err

    while not results.empty():
        _parse(results)



if __name__ == '__main__':
    main()

利用Python3中的concurrent.futures包

在Python3中能夠利用concurrent.futures包,編寫更加簡單易用的多線程/多進程代碼.其使用感受和Java的concurrent框架很類似(借鑑?)
好比下面的簡單代碼示例

pythondef handler():
    futures = set()

    with concurrent.futures.ProcessPoolExecutor(max_workers=cpu_count) as executor:
        for task in get_task(tasks):
            future = executor.submit(task)
            futures.add(future)

def wait_for(futures):
    try:
        for future in concurrent.futures.as_completed(futures):
            err = futures.exception()
            if not err:
                result = future.result()
            else:
                raise err
    except KeyboardInterrupt as e:
        for future in futures:
            future.cancel()
        print "Task has been canceled!"
        print e
    return result

總結

要是一些大型Python項目也這般編寫,那麼效率也過低了.在Python中有許多已有的框架使用,使用它們起來更加高效.
可是本身的一些"小打小鬧"的程序這樣來編寫仍是不錯的.:)