Python多進程解決方案multiprocessing ProcessPoolExecutor

 

大多數編程語言都會有多線程和多進程的概念，至於線程和進程的概念，大家可以百度一下。

作為一門膠水語言，Python毫不意外，也可以利用多線程和多進程處理並發問題，但是多線程由於GIL的存在，起作用范圍大打折扣，僅限於在IO等場景可以發揮點作用。

所以，今天要跟大家分享的是Python多進程方案，更好地利用系統多核，從而提升性能。

 

基礎方案一：

利用Process新建一個子進程，在子進程執行任務。我們

寫一個循環，模擬耗時任務的執行。主進程不會等待子進程執行完，就會繼續往下執行。我們可以根據實際業務情況，開啟很多這樣的進程。

# encoding=utf-8
from datetime import datetime
import time
from multiprocessing import Pool, Process


def job1():
    def run():
        print('[Child][{}]'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')))

    # 模擬一個耗時任務
    counter = 0
    while counter < 5:
        run()
        time.sleep(2)
        counter += 1
    #
    print('sub process done')


if __name__ == '__main__':
    p = Process(target=job1)
    p.start()
    #
    print('main process done')

執行效果如下

 

 

 

方案二：進程池

進程池可以理解成一個隊列，該隊列可以容易指定數量的子進程，當隊列被任務占滿之后，后續新增的任務就得排隊，直到舊的進程有任務執行完空余出來，才會去執行新的任務。

# encoding=utf-8
from datetime import datetime
import time
import random
from multiprocessing import Pool, Process


def job2(msg):
    def run():
        print('[Child-{}][{}]'.format(msg, datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')))

    # 模擬一個耗時任務
    time.sleep(random.randint(1, 3))
    run()


if __name__ == '__main__':
    # 進程池大小
    pool_size = 2
    # 進程池
    pool = Pool(pool_size)
    # 添加任務, 假設我們要添加6個任務，由於進程池大小為2，每次能只有2個任務並行執行，其他任務排隊
    [pool.apply_async(job2, args=(i,)) for i in range(6)]
    pool.close()
    pool.join()
    #
    print('main process done')

執行結果如下：

 

 

 

方案三：ProcessPoolExecutor

ProcessPoolExecutor是concurrent.futures里面的一個多進程解決方案，對多進程進行了一些便利的封裝，用起來跟Java的ThreadPoolExecutor感覺很類似。

同樣，方案二的示例，我們用ProcessPoolExecutor重新實現一次。

# encoding=utf-8
from datetime import datetime
import time
import random
from concurrent.futures import ProcessPoolExecutor, wait


def job3(msg):
    def run():
        print('[Child-{}][{}]'.format(msg, datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')))

    # 模擬一個耗時任務
    time.sleep(random.randint(1, 5))
    run()


if __name__ == '__main__':
    # 進程池大小
    pool_size = 2
    # 進程池
    pool = ProcessPoolExecutor(pool_size)
    # 添加任務, 假設我們要添加6個任務，由於進程池大小為2，每次能只有2個任務並行執行，其他任務排隊
    tasks = [pool.submit(job3, i) for i in range(6)]
    ### 等待任務執行完, 也可以設置一個timeout時間
    wait(tasks)
    #
    print('main process done')

 

 

 

結語

以上簡單介紹了多進程，進程池的幾種用法。當然示例的幾種場景都不涉及到多進程並發變量共享的問題，這個后續會推出其他文章單獨講述。