asyncio ではありませんが threading.Thread を使って want_item_url の実装をマルチスレッドにしたサンプルです。
import threading
# 省略
def want_item_url(links, name, color):
lock = threading.Lock()
target_url = []
def search_item(url, name, color):
item_page = requests.get(url)
soup = BeautifulSoup(item_page.content, 'lxml')
item_name = soup.select('h1[itemprop="name"]')[0].string
print(item_name)
item_color = soup.select('#details > p.style')[0].string
print(item_color)
if name in item_name and color in item_color:
with lock:
target_url.append(url)
threads = []
for i, link in enumerate(links):
url = 'https://www.supremenewyork.com' + link
th = threading.Thread(target=search_item, name="search_"+str(i), args=(url, name, color))
th.daemon = True
th.start()
threads.append(th)
for th in threads:
th.join()
return target_url
links = get_item_urls('accessories')
want_url = want_item_url(links, 'Crew Socks', 'White')
print(*want_url)
以下追記
links の 個々の requests.get() までの処理にして速度比較をした結果(単位:秒)では
Thread も asyncio もほぼ同じ(Thread が asyncio よりほんの少し速い)結果でした。
asyncio に関しては使用したことがないため want_url を書き込む際に排他が必要かどうか
から調べないといけませんので他の方の回答をご期待ください。
Thread 版 平均 0.166sec
0.213, 0.205, 0.147, 0.144, 0.157, 0.203, 0.154, 0.138, 0.148, 0.151
asyncio 版 平均 0.218sec
0.221, 0.203, 0.213, 0.246, 0.216, 0.230, 0.219, 0.218, 0.203, 0.210
Thread 版(性能比較用:search_item() は requests.get(url) のみ)
import requests
from bs4 import BeautifulSoup
import threading
import time
def get_item_urls(category):
url = 'https://www.supremenewyork.com/shop/all/' + category
category_page = requests.get(url)
soup = BeautifulSoup(category_page.content, 'lxml')
items_div = soup.select('article > .inner-article > a')
return [url.get('href') for url in items_div]
def want_item_url(links, name, color):
lock = threading.Lock()
target_url = []
def search_item(url, name, color):
item_page = requests.get(url)
threads = []
for i, link in enumerate(links):
url = 'https://www.supremenewyork.com' + link
th = threading.Thread(target=search_item, name="search_"+str(i), args=(url, name, color))
th.daemon = True
th.start()
threads.append(th)
for th in threads:
th.join()
return target_url
daemon = True
links = get_item_urls('accessories')
start = time.time()
want_url = want_item_url(links, 'Crew Socks', 'White')
print(time.time() - start)
asyncio 版
import asyncio
import requests
from bs4 import BeautifulSoup
import time
def get_item_urls(category):
url = 'https://www.supremenewyork.com/shop/all/' + category
category_page = requests.get(url)
soup = BeautifulSoup(category_page.content, 'lxml')
items_div = soup.select('article > .inner-article > a')
return [url.get('href') for url in items_div]
links = get_item_urls('accessories')
def ex(i):
r = requests.get('https://www.supremenewyork.com' + links[i])
return len(r.content)
async def handler(loop):
async def async_ex(i):
async with asyncio.Semaphore(20): # BUG! 2回目の追記コード参照
return await loop.run_in_executor(None, ex, i)
tasks = [async_ex(i) for i in range(len(links))]
return await asyncio.gather(*tasks)
start = time.time()
loop = asyncio.get_event_loop()
data = loop.run_until_complete(handler(loop))
print(time.time() - start)
以下2回目の追記です。
asyncio を使用して実装してみました。
非同期処理結果をすべて受け取れるため使いやすいですね。
性能を比較するとなぜか Thread 版の方がほんの少し速いですね。
Thread はリソースをそれなりに消費します。
アクセスする URL が数十であれば Thread でいいでしょうが多い場合は
asyncio を使うべきですね。
asyncio 要求機能実装版
import asyncio
import requests
from bs4 import BeautifulSoup
import time
def get_item_urls(category):
url = 'https://www.supremenewyork.com/shop/all/' + category
category_page = requests.get(url)
soup = BeautifulSoup(category_page.content, 'lxml')
items_div = soup.select('article > .inner-article > a')
return [url.get('href') for url in items_div]
def search_item(link, name, color):
url = 'https://www.supremenewyork.com' + link
item_page = requests.get(url)
soup = BeautifulSoup(item_page.content, 'lxml')
item_name = soup.select('h1[itemprop="name"]')[0].string
print(item_name)
item_color = soup.select('#details > p.style')[0].string
print(item_color)
if name in item_name and color in item_color:
return url
async def want_item_url(loop, links, name, color):
# 2020/9/5 以下4行修正
sem = asyncio.Semaphore(20) # 20: concurrency limitation 追加
async def async_ex(i):
#async with asyncio.Semaphore(20): # BUG! 削除
async with sem: # 修正
return await loop.run_in_executor(None, search_item, links[i], name, color)
tasks = [async_ex(i) for i in range(len(links))]
return await asyncio.gather(*tasks)
links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
data = [s for s in data if s]
loop.close()
print(*data)
print(time.time() - start)
9/5質問への回答
これがなぜ必要なのか分からない。
async def async_ex(i):
タスクの同時動作数を制限しないのであれば、次のように書けますね。
async def want_item_url(loop, links, name, color):
tasks = [loop.run_in_executor(None, search_item, links[i], name, color) for i in range(len(links))]
return await asyncio.gather(*tasks)
これでスレッドを制限している。なしでも可能か?
async with asyncio.Semaphore(20):
スレッドを制限してはいません。同時動作タスク数を制限するつもりのコードでした。
(同時動作タスク数を制限できていないコードでした。)
このコードでは、毎回セマフォオブジェクトを生成していますので意味がありません。
(1つのセマフォオブジェクトを共有して初めて意味がある)
バグのあるコードを提示してすみませんでした。修正内容を下に提示します。
(2回目の追記コードも修正しました)
asyncio はイベントループ上でタスクを切り替えています。スレッドを切り替えているわけではありません。
リファレンスでは、イベントループ関連のほとんどがスレッドアンセーフとなっています。
■修正前
async def want_item_url(loop, links, name, color):
async def async_ex(i):
async with asyncio.Semaphore(20):
■修正後
async def want_item_url(loop, links, name, color):
sem = asyncio.Semaphore(20) # 20: concurrency limitation
async def async_ex(i):
async with sem:
ここの関数には非同期処理をしたい関数を代入する。search_itemの関数自体をたくさんのスレッドで動作させている。
search_itemの一連の処理が済んだ順でこのコードだとurlがあったら処理を終了するのではなく、とりあえず全部のurlを巡回する仕様
仕様になっている。もし該当urlが見つかった瞬間にイベントループを抜けるにはどうしたら良いでしょうか?
もし該当 url が見つかった瞬間にイベントループを抜けるには、すでに動作している他のタスクをキャンセルする or 強制終了する等の面倒な処理が必要になるはずです。
リファレンスでは gather がキャンセルされた場合はキャンセルが伝搬されるとあるため gather のキャンセルで可能かもしれません。
再帰的になっているなぜ[]なのか返る値はリスト形式に入れられるのか?Futureがここに返ってくるの?
tasks = [async_ex(i) for i in range(len(links))]
勘違いされていませんか?再帰的にはなっていませんよ。
gtather渡したシーケンスの順番を保ってくれる。
return await asyncio.gather(*tasks)
リファレンスを見た限りでは、何も書かれていないようです。
https://docs.python.org/ja/3.6/library/asyncio-task.html#asyncio.gather
データの前にある*はなんなのか?
print(*data)
アスタリスクは、イテラブルアンパック演算子です。
この場合、リストをアンパックして print 関数の引数に渡しています。
| 非同期処理結果をすべて受け取れるため使いやすいですね。
これはtasks = [async_ex(i) for i in range(len(links))]に未完のfutureオブジェクト入って全て完了したら次の処理に移れる事を指しているのでしょうか?
「非同期処理結果をすべて受け取れる」と書いたのは
一番上のレベルで
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
と実行結果の各 URL を「戻り値として」すべて受け取れるの意味です。
Thread の場合は global か nonlocal の変数に各スレッドから排他処理して格納しなければなりません。
9/6 の追加質問の回答
gtatherのシーケンスを保つについてはこちらの記事に書かれていました。
こちらの記事には「シーケンスを保つ」ことのエビデンスが何も提示されていません。
実際の動作上は、確かに保たれていそうですが Python の言語リファレンス上では確認できませんでした。
先の async with asyncio.Semaphore(20): も Web の内容をうっかり使ってしまった結果です。
いつも Web 情報が信頼できるわけではありません。
イベントループを途中で抜けたくて
if name in item_name and color in item_color:
# ここにループを抜ける処理を書く
# task.cancel()を入れたら上手く動作する。
return url
task.cancel()を挟めばいいと思うのですが、肝心のtaskをどこから持ってきたら良いか分かりません。
もしご存知でしたら、教えて下さいお願いします。
そこで task.cancel() をして return url 以降がまともに動くのでしょうか?
どうしてもキャンセルしたいのであれば以下のようなアプローチでしょうかね。
asyncio.gather(*tasks) のオブジェクトとキャンセル用のフラグ(キャンセル未)と結果格納用の ans_url 変数をグローバルで保持し、
return url の代わりに「フラグをキャンセル可にし ans_url = url を実行」する。
search_item 関数内の
item_page = requests.get(url) ここが時間がかかる処理と認識しています
の直後で「フラグがキャンセル可であればフラグをキャンセル済みにし gather 結果オブジェクトを cancel() する」(キャンセル済みにすることで2重キャンセルを防止する)
その上で exception 処理を追加
try:
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
print("CancelledError", e)
等々が必要になるでしょうね。
その上で cancel() 後に CancelledError exception が発生する場合はタスクのリソースが正常に解放されるか心配になります。
キャンセルはコードが汚くなる上に気を付けることがいろいろでてきます。
また、キャンセルしたらすぐに loop.run_until_complete を中断してくれるかも気になります。(通信のレスポンスを待っているタスクをレスポンスが来ても問題ないようにしてすぐにキャンセルできるか?)
排他制御が不要なだけマルチスレッドのキャンセルよりは楽でしょう。
頑張ってみてください。
9/7 追記
グローバル変数でflagを作成したのですが実行すると見つからないとエラーが出ます。
global 宣言がされていないだけではありませんか?
Python における変数へのオブジェクトの代入文は、オブジェクトへの参照情報の代入です。
(リファレンス上は「束縛」という言葉を使っています)
つまり変数への代入は、オブジェクトへの参照先を変更することになります。
global スコープ以外で global 変数に代入以外のアクセスをするのは単にアクセス式を書く
だけですが、参照先を変更する代入を行う場合は global 宣言が必要です。
(代入前に代入以外のアクセスをする場合はその前に global 宣言が必要です。)
global と nonlocal の使い方は基本的な話です、把握しておきましょう。
以下、提示したアプローチによるサンプルコードです。
キャンセル処理に動きがわかるように print を入れています。
実際に動作させると時間のかかる requests.get(url) は並行して動作しているため一斉に終了します。
そのため requests.get(url) 直後に cancel() 処理を実行してもキャンセルすることはありませんでした。
soup.select() 後に入れることでキャンセルできますが、キャンセル処理を入れる効果は少ないですね。
こちらの環境では Thread 版の方が速いですよ。
20程度の URL アクセスで少しでも早くしたいのであれば Thread 版を検討されてはいかがでしょうか?
asincio はアプリケーションレベルでイベントループ内でタスク切り替えを行うため結構処理が重いようです。
(Thread は OS レベルで切り替えが行われます。)
なお、これまでのコードには loop.close() がなくお行儀が悪いコードでしたので、前の asyncio 要求機能実装版のコードも含め追記しました。
asyncio 要求機能実装版(Cancel 実装) 改善前
import asyncio
import requests
from bs4 import BeautifulSoup
import time
from enum import Enum
class TaskStatus(Enum):
NonCancelable = 0; Cancelable = 1; Canceled = 2
def get_item_urls(category):
url = 'https://www.supremenewyork.com/shop/all/' + category
category_page = requests.get(url)
soup = BeautifulSoup(category_page.content, 'lxml')
items_div = soup.select('article > .inner-article > a')
return [url.get('href') for url in items_div]
taskflag = TaskStatus.NonCancelable
ans_url = ""
def search_item(link, name, color):
global taskflag, ans_url
url = 'https://www.supremenewyork.com' + link
if taskflag != TaskStatus.NonCancelable:
if taskflag == TaskStatus.Cancelable:
gtask.cancel()
taskflag = TaskStatus.Canceled
print('*** gtask.cancel() prev requests.get(url) ***')
return
print('prev: ', time.time() - start)
item_page = requests.get(url)
print('after: ', time.time() - start)
if taskflag != TaskStatus.NonCancelable:
if taskflag == TaskStatus.Cancelable:
gtask.cancel()
taskflag = TaskStatus.Canceled
print('*** gtask.cancel() after requests.get(url) ***')
return
soup = BeautifulSoup(item_page.content, 'lxml')
item_name = soup.select('h1[itemprop="name"]')[0].string
print(item_name)
if taskflag != TaskStatus.NonCancelable:
if taskflag == TaskStatus.Cancelable:
gtask.cancel()
taskflag = TaskStatus.Canceled
print('*** gtask.cancel() after soup.select(h1) ***')
return
item_color = soup.select('#details > p.style')[0].string
print(item_color)
if taskflag != TaskStatus.NonCancelable:
if taskflag == TaskStatus.Cancelable:
gtask.cancel()
taskflag = TaskStatus.Canceled
print('*** gtask.cancel() after soup.select(details) ***')
return
if name in item_name and color in item_color:
if taskflag == TaskStatus.NonCancelable:
taskflag = TaskStatus.Cancelable
ans_url = url
print('*** set ans_url ***')
async def want_item_url(loop, links, name, color):
sem = asyncio.Semaphore(20) # 20: concurrency limitation
async def async_ex(i):
async with sem:
return await loop.run_in_executor(None, search_item, links[i], name, color)
tasks = [async_ex(i) for i in range(len(links))]
global gtask
gtask = asyncio.gather(*tasks)
return await gtask
links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
print("*** CancelledError ***", e)
finally:
loop.close()
print(ans_url)
print(time.time() - start)
item_page = requests.get(url) の前にもキャンセル処理追加 2020/09/07 12:21
上のコードでは gtask.cancel() を 3 ~ 4 回実行してしまっていました。
gtask.cancel() はタスク切り替えのイベントとなっているようです。
if taskflag == TaskStatus.Cancelable:
gtask.cancel()
taskflag = TaskStatus.Canceled
そのため本来であれば
if taskflag == TaskStatus.Cancelable:
taskflag = TaskStatus.Canceled
gtask.cancel()
と先にフラグを更新しないと多重キャンセルの対策になっていませんでした。
ただ、よく考えると
return url
をなくし
ans_url = url
とグローバルの ans_url に格納したので
ans_url = url
gtask.cancel()
とすることができました。この対応をした改善版を以下に提示します。
asyncio 要求機能実装版(Cancel 実装) 改善版
import asyncio
import requests
from bs4 import BeautifulSoup
import time
def get_item_urls(category):
url = 'https://www.supremenewyork.com/shop/all/' + category
category_page = requests.get(url)
soup = BeautifulSoup(category_page.content, 'lxml')
items_div = soup.select('article > .inner-article > a')
return [url.get('href') for url in items_div]
canceled = False
ans_url = ""
def search_item(link, name, color):
global canceled, ans_url
url = 'https://www.supremenewyork.com' + link
if canceled: return
item_page = requests.get(url)
if canceled: return
soup = BeautifulSoup(item_page.content, 'lxml')
item_name = soup.select('h1[itemprop="name"]')[0].string
print(item_name)
if canceled: return
item_color = soup.select('#details > p.style')[0].string
print(item_color)
if canceled: return
if name in item_name and color in item_color:
if not canceled:
canceled = True
ans_url = url
print('*** set ans_url ***')
gtask.cancel()
async def want_item_url(loop, links, name, color):
sem = asyncio.Semaphore(20) # 20: concurrency limitation
async def async_ex(i):
async with sem:
return await loop.run_in_executor(None, search_item, links[i], name, color)
tasks = [async_ex(i) for i in range(len(links))]
global gtask
gtask = asyncio.gather(*tasks)
return await gtask
links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
print("*** CancelledError ***", e)
finally:
loop.close()
print(ans_url)
print(time.time() - start)
上記コードの if canceled: return は必須ではありません。
gtask.cancel() 後、少しでも無駄な処理を省くために入れています。