並列・並行

★Asyncioの使い方

例１

import asyncio

import aiohttp

def download_many(url_list):

loop = asyncio.get_event_loop()

to_do = [download_one(url) for url in sorted(url_list)]

wait_coro = asyncio.wait(to_do)

res, _ = loop.run_until_complete(wait_coro)

loop.close()

return len(res)

async def download_one(url):

status, reason, contents = await request(url)

...

return ...

async def request(url):

async with aiohttp.ClientSession() as session:

return await fetch(session, url)

async def fetch(session, url):

async with session.get(url) as resp:

#return await resp.read()

return await response(resp)

async def response(resp):

return resp.status, resp.reason, await resp.read()

例２

tasks = []

urls = [...]

async def download_one(url):

async with aiohttp.ClientSession() as session:

async with session.get(url) as response:

# response = await response.read()

return await response.read()

def download_many():

for url in urls:

task = asyncio.ensure_future(download_one(url))

tasks.append(task)

if __name__ == '__main__':

loop = asyncio.get_event_loop()

download_many()

# loop.run_until_complete(asyncio.wait(tasks))

result = loop.run_until_complete(asyncio.gather(*tasks))

"too many file"エラー対策

例１

async def download_one(url, semaphore):

async with semaphore:

async with aiohttp.ClientSession() as session:

async with session.get(url) as response:

return await response.read()

async def download_many():

semaphore = asyncio.Semaphore(1000) # 制限した並列数1000

to_get = [download_one(url, semaphore) for url in urls]

await asyncio.wait(to_get)

if __name__ == '__main__':

loop = asyncio.get_event_loop()

loop.run_until_complete(download_many())

loop.close()

例２

import asyncio

from aiohttp import ClientSession

async def fetch(url, session):

# can not use connection pool if the different servers

# async with ClientSession() as session:

async with session.get(url) as response:

delay = response.headers.get("DELAY")

date = response.headers.get("DATE")

print("{}:{} with delay {}".format(date, response.url, delay))

return await response.read()

async def bound_fetch(sem, url, session):

async with sem:

await fetch(url, session)

async def run(r):

urls = [...]

tasks = []

sem = asyncio.Semaphore(1000)

# use connection pool if the same server

async with ClientSession() as session:

for url in urls:

task = asyncio.ensure_future(bound_fetch(sem, url, session))

tasks.append(task)

responses = asyncio.gather(*tasks)

await responses

number = 10000

loop = asyncio.get_event_loop()

future = asyncio.ensure_future(run(number))

loop.run_until_complete(future)

# loop.close()