首頁 > 軟體

python使用期物處理並行教學

2022-06-13 18:03:07

learning from 《流暢的python》

1. futures.ThreadPoolExecutor

import os
import time
import sys
import requests
POP20_CC = ('CN IN US ID BR PK NG BD RU JP ' 'MX PH VN ET EG DE IR TR CD FR').split()
BASE_URL = 'http://flupy.org/data/flags'
DEST_DIR = './'
def save_flag(img, filename):  # 儲存影象
    path = os.path.join(DEST_DIR, filename)
    with open(path, 'wb') as fp:
        fp.write(img)
def get_flag(cc):  # 獲取影象
    url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc=cc.lower())
    resp = requests.get(url)
    return resp.content
def show(text):  # 列印資訊
    print(text, end=' ')
    sys.stdout.flush()
def download_many(cc_list):
    for cc in sorted(cc_list):
        image = get_flag(cc)  # 獲取
        show(cc)  # 列印
        save_flag(image, cc.lower() + '.gif')  # 儲存
    return len(cc_list)
def main(download_many):
    t0 = time.time()
    count = download_many(POP20_CC)
    elapsed = time.time() - t0
    msg = 'n{} flags downloaded in {:.2f}s'
    print(msg.format(count, elapsed))  # 計時資訊
# ----使用 futures.ThreadPoolExecutor 類實現多執行緒下載
from concurrent import futures
MAX_WORKERS = 20  # 最多使用幾個執行緒
def download_one(cc):
    image = get_flag(cc)
    show(cc)
    save_flag(image, cc.lower() + '.gif')
    return cc
def download_many_1(cc_list):
    workers = min(MAX_WORKERS, len(cc_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        #  使用工作的執行緒數範例化 ThreadPoolExecutor 類;
        #  executor.__exit__ 方法會呼叫 executor.shutdown(wait=True) 方法,
        #  它會在所有執行緒都執行完畢 前阻塞執行緒
        res = executor.map(download_one, sorted(cc_list))
        # download_one 函數 會在多個執行緒中並行呼叫;
        # map 方法返回一個生成器,因此可以迭代, 獲取各個函數返回的值
    return len(list(res))
if __name__ == '__main__':
    # main(download_many) # 24 秒
    main(download_many_1)  # 3 秒

2. 期物

通常不應自己建立期物

只能由並行框架(concurrent.futures 或 asyncio)範例化 原因:期物 表示終將發生的事情,其 執行的時間 已經排定。因此,只有排定把某件事交給 concurrent.futures.Executor 子類處理時,才會建立 concurrent.futures.Future 範例

例如,Executor.submit() 方法的引數是一個可呼叫的物件,呼叫這個方法後會為傳入的可呼叫物件 排期,並返回一個期物

def download_many_2(cc_list):
    cc_list = cc_list[:5]
    with futures.ThreadPoolExecutor(max_workers=3) as executor:
        to_do = []
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc)
            # executor.submit 方法排定可呼叫物件的執行時間,
            # 然後返回一個 期物,表示這個待執行的操作
            to_do.append(future) # 儲存各個期物
            msg = 'Scheduled for {}: {}'
            print(msg.format(cc, future))
        results = []
        for future in futures.as_completed(to_do):
            # as_completed 函數在期物執行結束後產出期物
            res = future.result() # 獲取期物的結果
            msg = '{} result: {!r}'
            print(msg.format(future, res))
            results.append(res)
    return len(results)
輸出:
Scheduled for BR: <Future at 0x22da99d2d30 state=running>
Scheduled for CN: <Future at 0x22da99e1040 state=running>
Scheduled for ID: <Future at 0x22da99e1b20 state=running>
Scheduled for IN: <Future at 0x22da99ec520 state=pending>
Scheduled for US: <Future at 0x22da99ecd00 state=pending>
CN <Future at 0x22da99e1040 state=finished returned str> result: 'CN'
BR <Future at 0x22da99d2d30 state=finished returned str> result: 'BR'
ID <Future at 0x22da99e1b20 state=finished returned str> result: 'ID'
IN <Future at 0x22da99ec520 state=finished returned str> result: 'IN'
US <Future at 0x22da99ecd00 state=finished returned str> result: 'US'
5 flags downloaded in 3.20s

3. 阻塞型I/O和GIL

CPython 直譯器本身就不是執行緒安全的,因此有全域性直譯器鎖(GIL), 一次只允許使用一個執行緒執行 Python 位元組碼。因此,一個 Python 程序 通常不能同時使用多個 CPU 核心

標準庫中所有執行阻塞型 I/O 操作的函數,在等待作業系統返回結果時 都會釋放 GIL。 這意味著在 Python 語言這個層次上可以使用多執行緒,而 I/O 密集型 Python 程式能從中受益:一個 Python 執行緒等待網路響應時,阻塞型 I/O 函數會釋放 GIL,再執行一個執行緒(網路下載,檔案讀寫都屬於 IO 密集型)

4. 使用concurrent.futures模組啟動程序

這個模組實現的是真正 的平行計算,因為它使用 ProcessPoolExecutor 類把工作分配給多個 Python 程序處理。 因此,如果需要做 CPU 密集型處理,使用這個模組 能繞開 GIL,利用所有可用的 CPU 核心

點選檢視:程序、執行緒概念差異

使用 concurrent.futures 模組能特別輕鬆地 把 基於執行緒 的方案轉成 基於程序 的方案

ProcessPoolExecutor 的價值體現在 CPU 密集型 作業上

以上就是python使用期物處理並行教學的詳細內容,更多關於python期物處理並行的資料請關注it145.com其它相關文章!


IT145.com E-mail:sddin#qq.com