Fluent Python 第17章 futuresを使った並行処理 (Python 3.7版)

Fluent Python 第17章

「Fluent Python 17章 futuresを使った並行処理」の前半部分をPython 3.7で実装します。 公式のHPにもPython 3.7用のアップデートがあります。

1 共通項目

例 17.1 共通項目

import asyncio
import sys
import time
from concurrent import futures

import aiohttp
import nest_asyncio
import requests

nest_asyncio.apply()

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"

def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = requests.get(url)
    return resp.content

def show(text):
    sys.stdout.write(text + " ")
    sys.stdout.flush()

def main(func):
    t0 = time.time()
    count = func(POP20_CC)
    elapsed = time.time() - t0
    msg = "\n{} flags downloaded in {:.2f}s"
    print(msg.format(count, elapsed))

2 逐次型ダウンロードスクリプト

例 17.2 flags.py

def download_many(cc_list):
    for cc in sorted(cc_list):
        get_flag(cc)
        show(cc)

    return len(cc_list)

if __name__ == "__main__":
    main(download_many)

BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN 
20 flags downloaded in 0.43s

3 concurrent.futuresを使ったダウンロードスクリプト

例 17.3 flags_threadpool.py

MAX_WORKERS = 20

def download_one(cc):
    get_flag(cc)
    show(cc)
    return cc

def download_many_threadpool(cc_list):
    workers = min(MAX_WORKERS, len(cc_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        res = executor.map(download_one, sorted(cc_list))

    return len(list(res))

if __name__ == "__main__":
    main(download_many_threadpool)

BD BR DE CN CD ET FR ID EG IN IR JP MX PK RU US NG PH TR VN 
20 flags downloaded in 0.13s

例 17.4 flags_threadpool_ac.py

def download_many_threadpool_ac(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)
            to_do.append(future)
            msg = "Scheduled for {}: {}"
            print(msg.format(cc, future))

        results = []
        for future in futures.as_completed(to_do):
            res = future.result()
            msg = "{} result: {!r}"
            print(msg.format(future, res))
            results.append(res)

    return len(results)

if __name__ == "__main__":
    main(download_many_threadpool_ac)

Scheduled for BR: <Future at 0x272fbbc6208 state=running>
Scheduled for CN: <Future at 0x272fbbc6668 state=running>
Scheduled for ID: <Future at 0x272fbbbbb38 state=running>
Scheduled for IN: <Future at 0x272fbba0358 state=pending>
Scheduled for US: <Future at 0x272fbba00f0 state=pending>
CN BR ID <Future at 0x272fbbc6208 state=finished returned str> result: 'BR'
<Future at 0x272fbbbbb38 state=finished returned str> result: 'ID'
<Future at 0x272fbbc6668 state=finished returned str> result: 'CN'
IN <Future at 0x272fbba0358 state=finished returned str> result: 'IN'
US <Future at 0x272fbba00f0 state=finished returned str> result: 'US'

5 flags downloaded in 0.07s

4 asyncio/aiohttpを使ったダウンロードスクリプト

例 17.5 flags_asyncio.py

async def get_flag_async(session, cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    async with session.get(url) as resp:
        return await resp.read()

async def download_one_async(session, cc):
    await get_flag_async(session, cc)
    show(cc)
    return cc

async def download_many_async(cc_list):
    async with aiohttp.ClientSession() as session:
        aws = [
            asyncio.create_task(download_one_async(session, cc))
            for cc in sorted(cc_list)
        ]
        res = await asyncio.gather(*aws)

    return len(res)

if __name__ == "__main__":
    main(lambda x: asyncio.run(download_many_async(x)))

DE BD BR NG ID RU EG MX IN TR CN CD ET IR PH JP US VN PK FR 
20 flags downloaded in 0.06s