python 進程池、線程池 與異步調用、回調機制

進程池、線程池使用案例

from concurrent.futures import ProcessPoolExecutor  # 進程池模塊
from concurrent.futures import ThreadPoolExecutor  # 線程池模塊
import os, time, random

#  下面是以進程池為例， 線程池只是模塊改一下即可
def talk(name):
    print('name: %s  pis%s  run' % (name,os.getpid()))
    time.sleep(random.randint(1, 3))

if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)  # 設置線程池大小，默認等於cpu核數
    for i in range(10):
        pool.submit(talk, '進程%s' % i)  # 異步提交（只是提交需要運行的線程不等待）

    # 作用1：關閉進程池入口不能再提交了   作用2：相當於jion 等待進程池全部運行完畢
    pool.shutdown(wait=True)  
    print('主進程')

異步調用與同步調用

concurrent.futures模塊提供了高度封裝的異步調用接口
ThreadPoolExecutor：線程池，提供異步調用
ProcessPoolExecutor: 進程池，提供異步調用

• 同步調用

from concurrent.futures import ProcessPoolExecutor  # 進程池模塊
import os, time, random


# 1、同步調用： 提交完任務后、就原地等待任務執行完畢，拿到結果，再執行下一行代碼（導致程序串行執行）
def talk(name):
    print('name: %s  pis%s  run' % (name,os.getpid()))
    time.sleep(random.randint(1, 3))

if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)
    for i in range(10):
        pool.submit(talk, '進程%s' % i).result()  # 同步迪奧用，result()，相當於join 串行

    pool.shutdown(wait=True)
    print('主進程')

---

• 異步調用

from concurrent.futures import ProcessPoolExecutor  # 進程池模塊
import os, time, random

def talk(name):
    print('name: %s  pis%s  run' % (name,os.getpid()))
    time.sleep(random.randint(1, 3))

if __name__ == '__main__':
    pool = ProcessPoolExecutor(4)
    for i in range(10):
        pool.submit(talk, '進程%s' % i)  # 異步調用，不需要等待

    pool.shutdown(wait=True)
    print('主進程')

回調機制

可以為進程池或線程池內的每個進程或線程綁定一個函數，該函數在進程或線程的任務執行完畢后自動觸發，並接收任務的返回值當作參數，該函數稱為回調函數

#parse_page拿到的是一個future對象obj，需要用obj.result()拿到結果
p.submit(這里異步調用).add_done_callback(方法)

---

• 案例：下載解析網頁頁面

import time
import requests
from concurrent.futures import ThreadPoolExecutor  # 線程池模塊

def get(url):
    print('GET %s' % url)
    response = requests.get(url)  # 下載頁面
    time.sleep(3)  # 模擬網絡延時
    return {'url': url, 'content': response.text}  # 頁面地址和頁面內容

def parse(res):
    res = res.result()  # !取到res結果 【回調函數】帶參數需要這樣
    print('%s res is %s' % (res['url'], len(res['content'])))

if __name__ == '__main__':
    urls = {
        'http://www.baidu.com',
        'http://www.360.com',
        'http://www.iqiyi.com'
    }

    pool = ThreadPoolExecutor(2)
    for i in urls:
        pool.submit(get, i).add_done_callback(parse)  # 【回調函數】執行完線程后，跟一個函數 

模塊介紹

concurrent.futures模塊提供了高度封裝的異步調用接口
ThreadPoolExecutor：線程池，提供異步調用
ProcessPoolExecutor: 進程池，提供異步調用
Both implement the same interface, which is defined by the abstract Executor class.

基本方法

#submit(fn, *args, **kwargs)
異步提交任務

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循環submit的操作

#shutdown(wait=True) 
相當於進程池的pool.close()+pool.join()操作
wait=True，等待池內所有任務執行完畢回收完資源后才繼續
wait=False，立即返回，並不會等待池內的任務執行完畢
但不管wait參數為何值，整個程序都會等到所有任務執行完畢
submit和map必須在shutdown之前

#result(timeout=None)
取得結果

#add_done_callback(fn)
回調函數

使用方法示例

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

import requests

# 給線程池設置最大工作的線程數
pool = ThreadPoolExecutor(10)
urls = [
    ‘http://cms-bucket.nosdn.127.net/2018/10/16/10e36050547445f6b8972daf7373a22‘
    ‘2.jpeg‘,
    ‘http://kwcdn.000dn.com/swfs/59/39972xmyj0206/pm.jpg‘,
    ‘http://pic-bucket.nosdn.127.net/photo/0008/2018-10-14/DU48KHUS2FKJ0008NOS‘
    ‘.jpg‘,
    ‘http://cms-bucket.nosdn.127.net/2018/10/16/b3a3fab2d65a41b79e0764727ae6d17‘
    ‘9.jpeg‘,
    ‘http://cms-bucket.nosdn.127.net/2018/10/15/92cbe61fc5ec40ab94f5d2f0ed86771‘
    ‘8.jpeg‘,
]


def task(url):
    response = requests.get(url)
    return response


# response會傳遞到call_back的參數中
def call_back(response):
    # 拿到的是一個future的對象
    # 從對象中取出task中返回的結果
    response.result()
    # 對回調過來的信息進行解析
    pass


for url in urls:
    # 使用回調函數能夠用返回的結果實時的去解析, 實現異步非阻塞
    pool.submit(task, url).add_done_callback(call_back)

# 如果使用shutdown會等待所有任務執行結束后再去執行主線程中的代碼
# pool.shutdown(True)