在Python編碼中我們經(jīng)常討論的一個(gè)方面就是如何優(yōu)化模擬執(zhí)行的性能。盡管在考慮量化代碼時(shí)NumPy、SciPy和pandas在這方面已然非常有用，但在構(gòu)建事件驅(qū)動(dòng)系統(tǒng)時(shí)我們無(wú)法有效地使用這些工具。有沒(méi)有可以加速我們代碼的其他辦法？答案是肯定的，但需要留意！

在這篇文章中，我們看一種不同的模型-并發(fā)，我們可以將它引入我們Python程序中。這種模型在模擬中工作地特別好，它不需要共享狀態(tài)。Monte Carlo模擬器可以用來(lái)做期權(quán)定價(jià)以及檢驗(yàn)算法交易等類型的各種參數(shù)的模擬。

我們將特別考慮Threading庫(kù)和Multiprocessing庫(kù)。

Python并發(fā)

當(dāng)Python初學(xué)者探索多線程的代碼為了計(jì)算密集型優(yōu)化時(shí)，問(wèn)得最多的問(wèn)題之一是：”當(dāng)我用多線程的時(shí)候，為什么我的程序變慢了？“

在多核機(jī)器上，我們期望多線程的代碼使用額外的核，從而提高整體性能。不幸的是，主Python解釋器（CPython）的內(nèi)部并不是真正的多線程，是通過(guò)一個(gè)全局解釋鎖（GIL）來(lái)進(jìn)行處理的。

GIL是必須的，因?yàn)镻ython解釋器是非線程安全的。這意味著當(dāng)從線程內(nèi)嘗試安全的訪問(wèn)Python對(duì)象的時(shí)候?qū)⒂幸粋€(gè)全局的強(qiáng)制鎖。在任何時(shí)候，僅僅一個(gè)單一的線程能夠獲取Python對(duì)象或者C API。每100個(gè)字節(jié)的Python指令解釋器將重新獲取鎖，這（潛在的）阻塞了I/0操作。因?yàn)殒i，CPU密集型的代碼使用線程庫(kù)時(shí)，不會(huì)獲得性能的提高，但是當(dāng)它使用多處理庫(kù)時(shí)，性能可以獲得提高。

并行庫(kù)的實(shí)現(xiàn)

現(xiàn)在，我們將使用上面所提到的兩個(gè)庫(kù)來(lái)實(shí)現(xiàn)對(duì)一個(gè)“小”問(wèn)題進(jìn)行并發(fā)優(yōu)化。

線程庫(kù)

上面我們提到： 運(yùn)行CPython解釋器的Python不會(huì)支持通過(guò)多線程來(lái)實(shí)現(xiàn)多核處理。不過(guò)，Python確實(shí)有一個(gè)線程庫(kù)。那么如果我們（可能）不能使用多個(gè)核心進(jìn)行處理，那么使用這個(gè)庫(kù)能取得什么好處呢？

許多程序，尤其是那些與網(wǎng)絡(luò)通信或者數(shù)據(jù)輸入/輸出(I/O)相關(guān)的程序，都經(jīng)常受到網(wǎng)絡(luò)性能或者輸入/輸出(I/O)性能的限制。這樣Python解釋器就會(huì)等待哪些從諸如網(wǎng)絡(luò)地址或者硬盤等“遠(yuǎn)端”數(shù)據(jù)源讀寫數(shù)據(jù)的函數(shù)調(diào)用返回。因此這樣的數(shù)據(jù)訪問(wèn)比從本地內(nèi)存或者CPU緩沖區(qū)讀取數(shù)據(jù)要慢的多。

因此，如果許多數(shù)據(jù)源都是通過(guò)這種方式訪問(wèn)的，那么就有一種方式對(duì)這種數(shù)據(jù)訪問(wèn)進(jìn)行性能提高，那就是對(duì)每個(gè)需要訪問(wèn)的數(shù)據(jù)項(xiàng)都產(chǎn)生一個(gè)線程 。

舉個(gè)例子，假設(shè)有一段Python代碼，它用來(lái)對(duì)許多站點(diǎn)的URL進(jìn)行扒取。再假定下載每個(gè)URL所需時(shí)間遠(yuǎn)遠(yuǎn)超過(guò)計(jì)算機(jī)CPU對(duì)它的處理時(shí)間，那么僅使用一個(gè)線程來(lái)實(shí)現(xiàn)就會(huì)大大地受到輸入/輸出(I/O)性能限制。

通過(guò)給每個(gè)下載資源生成一個(gè)新的線程，這段代碼就會(huì)并行地對(duì)多個(gè)數(shù)據(jù)源進(jìn)行下載，在所有下載都結(jié)束的時(shí)候再對(duì)結(jié)果進(jìn)行組合。這就意味著每個(gè)后續(xù)下載都不會(huì)等待前一個(gè)網(wǎng)頁(yè)下載完成。此時(shí)，這段代碼就受收到客戶/服務(wù)端帶寬的限制。

不過(guò)，許多與財(cái)務(wù)相關(guān)的應(yīng)用都受到CPU性能的限制，這是因?yàn)檫@樣的應(yīng)用都是高度集中式的對(duì)數(shù)字進(jìn)行處理。這樣的應(yīng)用都會(huì)進(jìn)行大型線性代數(shù)計(jì)算或者數(shù)值的隨機(jī)統(tǒng)計(jì)，比如進(jìn)行蒙地卡羅模擬統(tǒng)計(jì)。所以只要對(duì)這樣的應(yīng)用使用Python和全局解釋鎖(GIL)，此時(shí)使用Python線程庫(kù)就不會(huì)有任何性能的提高。

Python實(shí)現(xiàn)

下面這段依次添加數(shù)字到列表的“玩具”代碼，舉例說(shuō)明了多線程的實(shí)現(xiàn)。每個(gè)線程創(chuàng)建一個(gè)新的列表并隨機(jī)添加一些數(shù)字到列表中。這個(gè)已選的“玩具”例子對(duì)CPU的消耗非常高。

下面的代碼概述了線程庫(kù)的接口，但是他不會(huì)比我們用單線程實(shí)現(xiàn)的速度更快。當(dāng)我們對(duì)下面的代碼用多處理庫(kù)時(shí)，我們將看到它會(huì)顯著的降低總的運(yùn)行時(shí)間。

讓我們檢查一下代碼是怎樣工作的。首先我們導(dǎo)入threading庫(kù)。然后我們創(chuàng)建一個(gè)帶有三個(gè)參數(shù)的函數(shù)list_append。第一個(gè)參數(shù)count定義了創(chuàng)建列表的大小。第二個(gè)參數(shù)id是“工作”(用于我們輸出debug信息到控制臺(tái))的ID。第三個(gè)參數(shù)out_list是追加隨機(jī)數(shù)的列表。

__main__函數(shù)創(chuàng)建了一個(gè)107的size，并用兩個(gè)threads執(zhí)行工作。然后創(chuàng)建了一個(gè)jobs列表，用于存儲(chǔ)分離的線程。threading.Thread對(duì)象將list_append函數(shù)作為參數(shù)，并將它附加到j(luò)obs列表。

最后，jobs分別開始并分別“joined”。join()方法阻塞了調(diào)用的線程（例如主Python解釋器線程）直到線程終止。在打印完整的信息到控制臺(tái)之前，確認(rèn)所有的線程執(zhí)行完成。

# thread_test.pyimport randomimport threadingdef list_append(count, id, out_list):  
    """  
    Creates an empty list and then appends a   
    random number to the list 'count' number  
    of times. A CPU-heavy operation!  
    """ 
    for i in range(count):  
        out_list.append(random.random())if __name__ == "__main__":  
    size = 10000000   # Number of random numbers to add  
    threads = 2   # Number of threads to create  
 
    # Create a list of jobs and then iterate through  
    # the number of threads appending each thread to  
    # the job list   
    jobs = []  
    for i in range(0, threads):  
        out_list = list()  
        thread = threading.Thread(target=list_append(size, i, out_list))  
        jobs.append(thread)  
 
    # Start the threads (i.e. calculate the random number lists)  
    for j in jobs:  
        j.start()  
 
    # Ensure all of the threads have finished  
    for j in jobs:  
        j.join()  
 
    print "List processing complete." 

我們能在控制臺(tái)中調(diào)用如下的命令time這段代碼

time python thread_test.py 

將產(chǎn)生如下的輸出

List processing complete.  
real    0m2.003s 
user    0m1.838s 
sys     0m0.161s 

注意user時(shí)間和sys時(shí)間相加大致等于real時(shí)間。這表明我們使用線程庫(kù)沒(méi)有獲得性能的提升。我們期待real時(shí)間顯著的降低。在并發(fā)編程的這些概念中分別被稱為CPU時(shí)間和掛鐘時(shí)間（wall-clock time）

多進(jìn)程處理庫(kù)
 

為了充分地使用所有現(xiàn)代處理器所能提供的多個(gè)核心 ，我們就要使用多進(jìn)程處理庫(kù) 。它的工作方式與線程庫(kù)完全不同 ，不過(guò)兩種庫(kù)的語(yǔ)法卻非常相似 。

多進(jìn)程處理庫(kù)事實(shí)上對(duì)每個(gè)并行任務(wù)都會(huì)生成多個(gè)操作系統(tǒng)進(jìn)程。通過(guò)給每個(gè)進(jìn)程賦予單獨(dú)的Python解釋器和單獨(dú)的全局解釋鎖(GIL)十分巧妙地規(guī)避了一個(gè)全局解釋鎖所帶來(lái)的問(wèn)題。而且每個(gè)進(jìn)程還可獨(dú)自占有一個(gè)處理器核心，在所有進(jìn)程處理都結(jié)束的時(shí)候再對(duì)結(jié)果進(jìn)行重組。

不過(guò)也存在一些缺陷。生成許多進(jìn)程就會(huì)帶來(lái)很多I/O管理問(wèn)題，這是因?yàn)槎鄠€(gè)處理器對(duì)數(shù)據(jù)的處理會(huì)引起數(shù)據(jù)混亂 。這就會(huì)導(dǎo)致整個(gè)運(yùn)行時(shí)間增多 。不過(guò)，假設(shè)把數(shù)據(jù)限制在每個(gè)進(jìn)程內(nèi)部 ，那么就可能大大的提高性能 。當(dāng)然，再怎么提高也不會(huì)超過(guò)阿姆達(dá)爾法則所規(guī)定的極限值。

Python實(shí)現(xiàn)

使用Multiprocessing實(shí)現(xiàn)僅僅需要修改導(dǎo)入行和multiprocessing.Process行。這里單獨(dú)的向目標(biāo)函數(shù)傳參數(shù)。除了這些，代碼幾乎和使用Threading實(shí)現(xiàn)的一樣：

# multiproc_test.pyimport randomimport multiprocessingdef list_append(count, id, out_list):  
    """  
    Creates an empty list and then appends a   
    random number to the list 'count' number  
    of times. A CPU-heavy operation!  
    """ 
    for i in range(count):  
        out_list.append(random.random())if __name__ == "__main__":  
    size = 10000000   # Number of random numbers to add  
    procs = 2   # Number of processes to create  
 
    # Create a list of jobs and then iterate through  
    # the number of processes appending each process to  
    # the job list   
    jobs = []  
    for i in range(0, procs):  
        out_list = list()  
        process = multiprocessing.Process(target=list_append,   
                                          args=(size, i, out_list))  
        jobs.append(process)  
 
    # Start the processes (i.e. calculate the random number lists)        
    for j in jobs:  
        j.start()  
 
    # Ensure all of the processes have finished  
    for j in jobs:  
        j.join()  
 
    print "List processing complete." 

控制臺(tái)測(cè)試運(yùn)行時(shí)間：

time python multiproc_test.py 

得到如下輸出：

List processing complete.  
real    0m1.045s 
user    0m1.824s 
sys     0m0.231s 

在這個(gè)例子中可以看到user和sys時(shí)間基本相同，而real下降了近兩倍。之所以會(huì)這樣是因?yàn)槲覀兪褂昧藘蓚€(gè)進(jìn)程。擴(kuò)展到四個(gè)進(jìn)程或者將列表的長(zhǎng)度減半結(jié)果如下（假設(shè)你的電腦至少是四核的）：

List processing complete.  
real    0m0.540s 
user    0m1.792s 
sys     0m0.269s 

使用四個(gè)進(jìn)程差不多提高了3.8倍速度。但是，在將這個(gè)規(guī)律推廣到更大范圍，更復(fù)雜的程序上時(shí)要小心。數(shù)據(jù)轉(zhuǎn)換，硬件cacha層次以及其他一些問(wèn)題會(huì)減弱加快的速度。

在下一篇文章中我們會(huì)將Event-Driben Basketer并行化，從而提高其運(yùn)行多維參數(shù)尋優(yōu)的能力。

相關(guān)閱讀：

• Cholesky Decomposition in Python and NumPy

• European Vanilla Call-Put Option Pricing with Python

• Jacobi Method in Python and NumPy

• LU Decomposition in Python and NumPy

• Options Pricing in Python

• QR Decomposition with Python and NumPy

• Quick-Start Python Quantitative Research Environment on Ubuntu 14.04

英文原文：Parallelising Python with Threading and Multiprocessing

譯文鏈接：http://www.oschina.net/translate/parallelising-python-with-threading-and-multiprocessing