問題起因

最近要將一個文本分割成好幾個topic，每個topic設(shè)計一個regressor，各regressor是相互獨立的，最后匯總所有topic的regressor得到總得預測結(jié)果。沒錯！類似bagging ensemble！只是我沒有抽樣。文本不大，大概3000行，topic個數(shù)為8，于是我寫了一個串行的程序，一個topic算完之后再算另一個topic?？墒俏以诿總€topic中用了GridSearchCV來調(diào)參，又要選特征又要調(diào)整regressor的參數(shù)，導致參數(shù)組合一共有1782種。我真是低估了調(diào)參的時間，程序跑了一天一夜最后因為忘記import一個庫導致最終的預測精度沒有算出來。后來想到，既然每個topic的預測都是獨立的，那是不是可以并行呢？

Python中的多線程與多進程

但是聽聞Python的多線程實際上并不能真正利用多核，所以如果使用多線程實際上還是在一個核上做并發(fā)處理。不過，如果使用多進程就可以真正利用多核，因為各進程之間是相互獨立的，不共享資源，可以在不同的核上執(zhí)行不同的進程，達到并行的效果。同時在我的問題中，各topic相互獨立，不涉及進程間的通信，只需最后匯總結(jié)果，因此使用多進程是個不錯的選擇。

multiprocessing

一個子進程

multiprocessing模塊提供process類實現(xiàn)新建進程。下述代碼是新建一個子進程。

from multiprocessing import Process

def f(name):
  print 'hello', name

if __name__ == '__main__':
  p = Process(target=f, args=('bob',)) # 新建一個子進程p，目標函數(shù)是f，args是函數(shù)f的參數(shù)列表
  p.start() # 開始執(zhí)行進程
  p.join() # 等待子進程結(jié)束

上述代碼中p.join()的意思是等待子進程結(jié)束后才執(zhí)行后續(xù)的操作，一般用于進程間通信。例如有一個讀進程pw和一個寫進程pr，在調(diào)用pw之前需要先寫pr.join()，表示等待寫進程結(jié)束之后才開始執(zhí)行讀進程。

多個子進程

如果要同時創(chuàng)建多個子進程可以使用multiprocessing.Pool類。該類可以創(chuàng)建一個進程池，然后在多個核上執(zhí)行這些進程。

import multiprocessing
import time

def func(msg):
  print multiprocessing.current_process().name + '-' + msg

if __name__ == "__main__":
  pool = multiprocessing.Pool(processes=4) # 創(chuàng)建4個進程
  for i in xrange(10):
    msg = "hello %d" %(i)
    pool.apply_async(func, (msg, ))
  pool.close() # 關(guān)閉進程池，表示不能在往進程池中添加進程
  pool.join() # 等待進程池中的所有進程執(zhí)行完畢，必須在close()之后調(diào)用
  print "Sub-process(es) done."

輸出結(jié)果如下：

Sub-process(es) done.
PoolWorker-34-hello 1
PoolWorker-33-hello 0
PoolWorker-35-hello 2
PoolWorker-36-hello 3
PoolWorker-34-hello 7
PoolWorker-33-hello 4
PoolWorker-35-hello 5
PoolWorker-36-hello 6
PoolWorker-33-hello 8
PoolWorker-36-hello 9

上述代碼中的pool.apply_async()是apply()函數(shù)的變體，apply_async()是apply()的并行版本，apply()是apply_async()的阻塞版本，使用apply()主進程會被阻塞直到函數(shù)執(zhí)行結(jié)束，所以說是阻塞版本。apply()既是Pool的方法，也是Python內(nèi)置的函數(shù)，兩者等價?？梢钥吹捷敵鼋Y(jié)果并不是按照代碼for循環(huán)中的順序輸出的。

多個子進程并返回值

apply_async()本身就可以返回被進程調(diào)用的函數(shù)的返回值。上一個創(chuàng)建多個子進程的代碼中，如果在函數(shù)func中返回一個值，那么pool.apply_async(func, (msg, ))的結(jié)果就是返回pool中所有進程的值的對象（注意是對象，不是值本身）。

import multiprocessing
import time

def func(msg):
  return multiprocessing.current_process().name + '-' + msg

if __name__ == "__main__":
  pool = multiprocessing.Pool(processes=4) # 創(chuàng)建4個進程
  results = []
  for i in xrange(10):
    msg = "hello %d" %(i)
    results.append(pool.apply_async(func, (msg, )))
  pool.close() # 關(guān)閉進程池，表示不能再往進程池中添加進程，需要在join之前調(diào)用
  pool.join() # 等待進程池中的所有進程執(zhí)行完畢
  print ("Sub-process(es) done.")

  for res in results:
    print (res.get())

上述代碼輸出結(jié)果如下：

Sub-process(es) done.
PoolWorker-37-hello 0
PoolWorker-38-hello 1
PoolWorker-39-hello 2
PoolWorker-40-hello 3
PoolWorker-37-hello 4
PoolWorker-38-hello 5
PoolWorker-39-hello 6
PoolWorker-37-hello 7
PoolWorker-40-hello 8
PoolWorker-38-hello 9

與之前的輸出不同，這次的輸出是有序的。

如果電腦是八核，建立8個進程，在Ubuntu下輸入top命令再按下大鍵盤的1，可以看到每個CPU的使用率是比較平均的，如下圖：

在system monitor中也可以清楚看到執(zhí)行多進程前后CPU使用率曲線的差異。

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

最新評論