GIL
在Python中,由于歷史原因(GIL),使得Python中多線程的效果非常不理想.GIL使得任何時(shí)刻Python只能利用一個(gè)CPU核,并且它的調(diào)度算法簡(jiǎn)單粗暴:多線程中,讓每個(gè)線程運(yùn)行一段時(shí)間t,然后強(qiáng)行掛起該線程,繼而去運(yùn)行其他線程,如此周而復(fù)始,直到所有線程結(jié)束.
這使得無(wú)法有效利用計(jì)算機(jī)系統(tǒng)中的"局部性",頻繁的線程切換也對(duì)緩存不是很友好,造成資源的浪費(fèi).
據(jù)說(shuō)Python官方曾經(jīng)實(shí)現(xiàn)了一個(gè)去除GIL的Python解釋器,但是其效果還不如有GIL的解釋器,遂放棄.后來(lái)Python官方推出了"利用多進(jìn)程替代多線程"的方案,在Python3中也有concurrent.futures這樣的包,讓我們的程序編寫(xiě)可以做到"簡(jiǎn)單和性能兼得".
多進(jìn)程/多線程+Queue
一般來(lái)說(shuō),在Python中編寫(xiě)并發(fā)程序的經(jīng)驗(yàn)是:計(jì)算密集型任務(wù)使用多進(jìn)程,IO密集型任務(wù)使用多進(jìn)程或者多線程.另外,因?yàn)樯婕暗劫Y源共享,所以需要同步鎖等一系列麻煩的步驟,代碼編寫(xiě)不直觀.另外一種好的思路是利用多進(jìn)程/多線程+Queue的方法,可以避免加鎖這樣麻煩低效的方式.
現(xiàn)在在Python2中利用Queue+多進(jìn)程的方法來(lái)處理一個(gè)IO密集型任務(wù).
假設(shè)現(xiàn)在需要下載多個(gè)網(wǎng)頁(yè)內(nèi)容并進(jìn)行解析,單進(jìn)程的方式效率很低,所以使用多進(jìn)程/多線程勢(shì)在必行.
我們可以先初始化一個(gè)tasks隊(duì)列,里面將要存儲(chǔ)的是一系列dest_url,同時(shí)開(kāi)啟4個(gè)進(jìn)程向tasks中取任務(wù)然后執(zhí)行,處理結(jié)果存儲(chǔ)在一個(gè)results隊(duì)列中,最后對(duì)results中的結(jié)果進(jìn)行解析.最后關(guān)閉兩個(gè)隊(duì)列.
下面是一些主要的邏輯代碼.
# -*- coding:utf-8 -*-#IO密集型任務(wù)#多個(gè)進(jìn)程同時(shí)下載多個(gè)網(wǎng)頁(yè)#利用Queue+多進(jìn)程#由于是IO密集型,所以同樣可以利用threading模塊import multiprocessingdef main(): tasks = multiprocessing.JoinableQueue() results = multiprocessing.Queue() cpu_count = multiprocessing.cpu_count() #進(jìn)程數(shù)目==CPU核數(shù)目 create_process(tasks, results, cpu_count) #主進(jìn)程馬上創(chuàng)建一系列進(jìn)程,但是由于阻塞隊(duì)列tasks開(kāi)始為空,副進(jìn)程全部被阻塞 add_tasks(tasks) #開(kāi)始往tasks中添加任務(wù) parse(tasks, results) #最后主進(jìn)程等待其他線程處理完成結(jié)果def create_process(tasks, results, cpu_count): for _ in range(cpu_count): p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根據(jù)_worker創(chuàng)建對(duì)應(yīng)的進(jìn)程 p.daemon = True #讓所有進(jìn)程可以隨主進(jìn)程結(jié)束而結(jié)束 p.start() #啟動(dòng)def _worker(tasks, results): while True: #因?yàn)榍懊嫠芯€程都設(shè)置了daemon=True,故不會(huì)無(wú)限循環(huán) try: task = tasks.get() #如果tasks中沒(méi)有任務(wù),則阻塞 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包,編寫(xiě)更加簡(jiǎn)單易用的多線程/多進(jìn)程代碼.其使用感覺(jué)和Java的concurrent框架很相似(借鑒?)
比如下面的簡(jiǎn)單代碼示例
def 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
總結(jié)
要是一些大型Python項(xiàng)目也這般編寫(xiě),那么效率也太低了.在Python中有許多已有的框架使用,使用它們起來(lái)更加高效.
但是自己的一些"小打小鬧"的程序這樣來(lái)編寫(xiě)還是不錯(cuò)的.:)
