Pythonでの並行処理：concurrent.futures徹底活用

IT・プログラミング

Pythonでの並行処理：concurrent.futures徹底活用

並行処理とは？なぜ`concurrent.futures`を使うのか

並行処理の基本：時間を有効活用するテクニック

並行処理とは、複数のタスクをあたかも同時に実行しているかのように見せる技術です。これは、タスクを細かく分割し、CPUがそれぞれのタスクを少しずつ処理していくことで実現されます。例えば、料理をするときに、複数の料理を同時進行で作るイメージです。一つの料理を切っている間に、別の料理を煮込むといった具合です。

並行処理と似た言葉に「並列処理」がありますが、これらは異なる概念です。並行処理はあくまで「同時進行 風」であり、実際にはCPUがタスクを切り替えながら実行します。一方、並列処理は、複数のCPUコアを使って、複数のタスクを 文字通り同時に 実行します。つまり、並列処理はより強力な同時実行の方法と言えます。

マルチスレッド vs マルチプロセス：処理の種類で使い分け

並行処理を実現する方法として、主に「マルチスレッド」と「マルチプロセス」の2つがあります。

  • マルチスレッド: 1つのプログラム（プロセス）の中で、複数の処理の流れ（スレッド）を同時に実行します。スレッドは同じメモリ空間を共有するため、データのやり取りが比較的簡単です。しかし、PythonにはGIL（Global Interpreter Lock）という制約があり、複数のスレッドが同時にPythonのコードを実行できません。そのため、CPUをたくさん使う処理（CPUバウンドな処理）では、あまり効果を発揮できません。主に、Webからのデータ取得やファイルへの書き込みなど、処理時間の大半がI/O待ちになるような処理（I/Oバウンドな処理）に向いています。
  • マルチプロセス: 複数のプログラム（プロセス）を同時に実行します。プロセスはそれぞれ独立したメモリ空間を持つため、データのやり取りには特別な仕組み（プロセス間通信）が必要です。マルチスレッドとは異なり、GILの制約を受けないため、CPUバウンドな処理を並行して実行するのに適しています。ただし、プロセスの作成には時間がかかるため、短時間で終わる処理をたくさん並行処理するのには向いていません。

`concurrent.futures`：並行処理をもっと手軽に

Pythonのconcurrent.futuresモジュールは、マルチスレッドとマルチプロセスを簡単に扱えるようにするための高レベルなインターフェースを提供します。このモジュールを使うことで、スレッドやプロセスの生成、管理、同期といった複雑な処理を意識せずに、並行処理を実装できます。

concurrent.futuresの主なメリットは以下の通りです。

  • シンプル: スレッドやプロセスの細かい設定を気にせず、簡単に並行処理を実装できます。
  • 柔軟性: ThreadPoolExecutor（スレッド）とProcessPoolExecutor（プロセス）を切り替えるだけで、処理の種類に最適な並行処理を選択できます。
  • 非同期処理のサポート: Futureオブジェクトを使って、非同期処理の結果を簡単に取得できます。

具体的な利用シーン：どんな時に使う？

concurrent.futuresは、以下のような場合に特に役立ちます。

  • Webスクレイピング: 複数のWebサイトから情報を収集する処理を並行して行うことで、大幅な時間短縮が期待できます。
  • 画像処理: 大量の画像を処理する際に、処理を分割して並行して行うことで、全体の処理時間を短縮できます。
  • データ分析: 大規模なデータセットに対する計算処理を並行して行うことで、分析時間を短縮できます。
  • Webアプリケーション: 複数のAPIリクエストを並行して行うことで、レスポンスタイムを改善できます。

まとめ：`concurrent.futures`で処理を効率化しよう！

並行処理は、プログラムの処理速度を向上させるための強力なテクニックです。concurrent.futuresモジュールを使うことで、マルチスレッドやマルチプロセスの複雑さを気にせずに、手軽に並行処理を実装できます。ぜひ、concurrent.futuresを活用して、日々の開発業務を効率化しましょう。

`ThreadPoolExecutor`：I/Oバウンドな処理を高速化

前のセクションでは、並行処理の基本とconcurrent.futuresを使うメリットについて解説しました。このセクションでは、concurrent.futuresモジュールの中でも、特にI/Oバウンドな処理の効率化に役立つThreadPoolExecutorに焦点を当てて解説します。WebリクエストやファイルI/Oなど、処理時間の多くがI/O待ちに費やされるタスクを、いかに高速化できるのかを見ていきましょう。

`ThreadPoolExecutor`とは？

ThreadPoolExecutorは、スレッドプールを管理し、与えられたタスクを複数のスレッドで並行して実行するためのクラスです。スレッドプールとは、あらかじめ用意されたスレッドの集合体のこと。タスクが来るたびにスレッドを生成・破棄するのではなく、プールから利用可能なスレッドを取り出してタスクを実行するため、オーバーヘッドを削減できます。

I/Oバウンドな処理とは、ネットワーク経由でのデータ取得（Webリクエストなど）や、ディスクからのファイル読み書きなど、CPUの計算能力よりも外部デバイスとの通信速度がボトルネックになる処理のことです。これらの処理は、CPUが処理を待機している時間が長いため、複数のタスクを並行して実行することで、全体の処理時間を短縮できます。

I/Oバウンドな処理の効率化テクニック

具体的な例を挙げながら、ThreadPoolExecutorを使ったI/Oバウンドな処理の効率化テクニックを見ていきましょう。

1. Webリクエストの並行処理

複数のWebサイトからデータを取得するWebスクレイピングを考えてみましょう。requestsライブラリを使ってWebサイトにリクエストを送信し、HTMLを取得する処理は、ネットワークの速度に依存するためI/Oバウンドです。ThreadPoolExecutorを使うことで、複数のWebサイトへのリクエストを並行して行うことができます。

“`python
import concurrent.futures
import requests

urls = [‘https://www.example.com’, ‘https://www.python.org’, ‘https://www.google.com’]

def fetch_url(url):
try:
response = requests.get(url, timeout=10)
response.raise_for_status() # HTTPエラーをチェック
return url, len(response.content)
except requests.exceptions.RequestException as e:
return url, str(e) # エラーメッセージを返す

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
for future in concurrent.futures.as_completed(future_to_url):
url, result = future.result()
print(f'{url}: {result}’)
“`

この例では、max_workers=3として、最大3つのスレッドで並行してWebリクエストを実行しています。executor.submit()でタスクを投入し、concurrent.futures.as_completed()で完了した順に結果を取得しています。

2. ファイルI/Oの並行処理

複数のファイルからデータを読み込む場合も、ThreadPoolExecutorが役立ちます。例えば、ログファイルを解析し、特定のエラーメッセージを検索する処理を考えてみましょう。ファイルサイズが大きい場合、ファイルI/Oがボトルネックになる可能性があります。

“`python
import concurrent.futures
import os

def process_file(filepath):
try:
with open(filepath, ‘r’) as f:
content = f.read()
# ここでcontentに対して何らかの処理を行う
return filepath, len(content)
except Exception as e:
return filepath, str(e)

file_paths = [‘file1.txt’, ‘file2.txt’, ‘file3.txt’] # サンプルのファイルパス

# サンプルファイルが存在しない場合に作成
for file in file_paths:
if not os.path.exists(file):
with open(file, ‘w’) as f:
f.write(“Sample content for ” + file)

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
future_to_file = {executor.submit(process_file, filepath): filepath for filepath in file_paths}
for future in concurrent.futures.as_completed(future_to_file):
filepath, result = future.result()
print(f'{filepath}: {result}’)
“`

この例では、process_file関数でファイルの読み込みと処理を行い、ThreadPoolExecutorで並行して実行しています。

`ThreadPoolExecutor`利用時の注意点

ThreadPoolExecutorは非常に強力なツールですが、利用にあたっていくつかの注意点があります。

  • GIL（Global Interpreter Lock）の影響: PythonのGILは、複数のスレッドが同時にPythonバイトコードを実行することを制限します。そのため、CPUバウンドな処理では、ThreadPoolExecutorを使っても劇的なパフォーマンス向上は期待できません。CPUバウンドな処理には、次のセクションで解説するProcessPoolExecutorが適しています。
  • スレッドセーフ: 複数のスレッドが共有リソースにアクセスする場合、競合状態が発生する可能性があります。ロックなどの同期プリミティブを使って、スレッドセーフなコードを記述する必要があります。
  • 例外処理: スレッド内で例外が発生した場合、そのままではプログラム全体が停止してしまう可能性があります。try...exceptブロックを使って、例外を適切に処理する必要があります。

まとめ

このセクションでは、ThreadPoolExecutorを使ってI/Oバウンドな処理を効率化する方法について解説しました。WebリクエストやファイルI/Oなど、I/O待ち時間が長い処理を並行して実行することで、全体の処理時間を大幅に短縮できます。ただし、GILの影響やスレッドセーフ、例外処理など、注意すべき点もいくつかあります。これらの点に注意しながら、ThreadPoolExecutorを効果的に活用してください。

次のセクションでは、CPUバウンドな処理の並列化に役立つProcessPoolExecutorについて解説します。

`ProcessPoolExecutor`：CPUバウンドな処理を並列化

「並行処理」と聞くと難しそう…と感じるかもしれませんが、concurrent.futuresを使えば、Pythonでも手軽に処理を高速化できます。特に、CPUをフル活用するような重たい処理（CPUバウンドな処理）では、ProcessPoolExecutorが威力を発揮します。今回は、ProcessPoolExecutorを使って、CPUバウンドな処理を並列化し、処理時間を短縮する方法を徹底解説します。

`ProcessPoolExecutor`とは？

ProcessPoolExecutorは、複数のプロセスを生成し、それらを活用してタスクを並列に実行するためのツールです。concurrent.futuresモジュールに含まれており、複雑なプロセス管理を隠蔽し、シンプルに並列処理を記述できるように設計されています。

なぜプロセスを使うのでしょうか？それは、PythonのGIL（Global Interpreter Lock）という仕組みが関係しています。GILは、複数のスレッドが同時にPythonのバイトコードを実行できないように制限するため、マルチスレッドではCPUバウンドな処理を十分に並列化できません。しかし、プロセスはそれぞれ独立したメモリ空間を持つため、GILの影響を受けずに複数のCPUコアをフル活用できるのです。

どんな時に`ProcessPoolExecutor`を使うべき？

ProcessPoolExecutorが活躍するのは、主に以下のようなCPUバウンドな処理です。

  • 数値計算：大規模な行列計算、統計処理、シミュレーションなど
  • 画像処理：画像のフィルタリング、変換、解析など
  • 動画処理：動画のエンコード、デコード、編集など
  • 機械学習：モデルの学習、推論など

これらの処理は、CPUパワーを大量に消費するため、並列化による恩恵を大きく受けることができます。

`ProcessPoolExecutor`を使った並列化の実例

具体的なコード例を見てみましょう。ここでは、簡単な数値計算（各数字の二乗和を計算する）をProcessPoolExecutorで並列化する例を紹介します。

“`python
import concurrent.futures
import time

def calculate_sum_of_squares(numbers):
total = 0
for number in numbers:
total += number * number
time.sleep(0.001) #CPU負荷をかけるため
return total

if __name__ == “__main__”:
numbers_list = [
list(range(1, 1001)),
list(range(1001, 2001)),
list(range(2001, 3001)),
list(range(3001, 4001)),
]

start_time = time.time()
with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
results = executor.map(calculate_sum_of_squares, numbers_list)

total_sum = sum(results)
end_time = time.time()

print(f”合計: {total_sum}”)
print(f”実行時間: {end_time – start_time:.2f}秒”)
“`

このコードでは、numbers_listの各リストに対してcalculate_sum_of_squares関数を並列に実行しています。executor.mapを使うことで、各リストがそれぞれのプロセスで処理され、最終的な合計が計算されます。

もし並列化せずに同じ処理を行うと、処理時間は大幅に増加します。ぜひ、実際にコードを実行して、並列化の効果を実感してみてください。

`ProcessPoolExecutor`を使う上での注意点

ProcessPoolExecutorを使う際には、以下の点に注意が必要です。

  • プロセスの起動オーバーヘッド: プロセスの生成には、スレッドよりも大きなオーバーヘッドがあります。そのため、処理時間が非常に短いタスクを並列化しても、オーバーヘッドの方が大きくなり、逆に遅くなる可能性があります。
  • プロセス間通信: プロセスは独立したメモリ空間を持つため、プロセス間でデータを共有するには、multiprocessingモジュールで提供されるキューやパイプなどのIPC（Inter-Process Communication）メカニズムを使用する必要があります。データの共有は、スレッド間よりも複雑になる場合があります。
  • デバッグ: 複数のプロセスが同時に実行されるため、デバッグが難しくなる場合があります。ログ出力を活用したり、デバッガを適切に設定したりする必要があります。

まとめ

ProcessPoolExecutorは、CPUバウンドな処理を並列化し、処理時間を大幅に短縮するための強力なツールです。GILの制限を回避し、複数のCPUコアを最大限に活用できます。数値計算、画像処理、動画処理など、CPUパワーを大量に消費する処理に積極的に活用しましょう。ただし、プロセスの起動オーバーヘッドやプロセス間通信の複雑さなど、注意点もいくつかあります。これらの点を理解した上で、ProcessPoolExecutorを効果的に活用し、Pythonコードのパフォーマンスを向上させましょう。

`Future`オブジェクト：非同期処理の結果を取得

並行処理の力を最大限に引き出すには、非同期処理の結果を効率的に取得することが不可欠です。ここで活躍するのがFutureオブジェクトです。Futureオブジェクトは、非同期的に実行されたタスクの結果を「未来」に取得するための、いわば引換券のようなものです。このセクションでは、Futureオブジェクトの基本から、エラーハンドリング、タイムアウト設定といった実践的なテクニックまでを徹底解説します。

`Future`オブジェクトとは？

Futureオブジェクトは、concurrent.futuresモジュールの中核をなす概念の一つです。Executor（ThreadPoolExecutorまたはProcessPoolExecutor）のsubmit()メソッドを使ってタスクを非同期に実行すると、そのタスクに対応するFutureオブジェクトが返されます。このFutureオブジェクトを通じて、タスクの実行状態を監視したり、結果を取得したり、例外を処理したりすることができます。

イメージとしては、レストランで注文した料理の引換券のようなものです。注文（タスクのsubmit()）すると引換券（Futureオブジェクト）が渡され、料理ができあがるまで（タスクの完了）待ちます。料理ができあがったら（タスクの結果が得られたら）、引換券を提示して料理を受け取る（Futureオブジェクトから結果を取得する）という流れです。

`Future`オブジェクトを使った結果の取得方法

Futureオブジェクトから結果を取得するには、主に以下の4つの方法があります。

  1. `result()`メソッド: このメソッドは、タスクが完了するまで処理をブロックし、タスクの結果を返します。もしタスクの実行中に例外が発生した場合、result()メソッドを呼び出すと、その例外が再送出されます。タイムアウトを設定することも可能です。例えば、future.result(timeout=10)とすると、10秒以内に結果が返ってこない場合はTimeoutErrorが発生します。

“`python
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(my_function, arg1, arg2)
try:
result = future.result(timeout=5) # 5秒のタイムアウトを設定
print(f”結果: {result}”)
except TimeoutError:
print(“タイムアウトしました”)
except Exception as e:
print(f”例外が発生しました: {e}”)
“`

  1. `exception()`メソッド: このメソッドは、タスクの実行中に例外が発生した場合、その例外オブジェクトを返します。例外が発生しなかった場合はNoneを返します。

“`python
exception = future.exception()
if exception:
print(f”例外が発生しました: {exception}”)
“`

  1. `add_done_callback()`メソッド: このメソッドを使うと、タスクが完了したときに自動的に呼び出されるコールバック関数を登録できます。コールバック関数はFutureオブジェクトを引数として受け取ります。

“`python
def done_callback(future):
try:
result = future.result()
print(f”タスク完了。結果: {result}”)
except Exception as e:
print(f”タスク完了。例外が発生しました: {e}”)

future.add_done_callback(done_callback)
“`

  1. `concurrent.futures.as_completed()`関数: 複数のFutureオブジェクトを監視し、完了した順にFutureオブジェクトを返すイテレータを生成します。複数のタスクの結果を、完了順に処理したい場合に便利です。

“`python
import concurrent.futures
import time

def task(n):
time.sleep(n)
return n

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(task, i) for i in range(1, 4)]
for future in concurrent.futures.as_completed(futures):
print(future.result())
“`

エラーハンドリングとタイムアウト設定

非同期処理では、エラーハンドリングとタイムアウト設定が特に重要になります。なぜなら、メインスレッドとは異なるスレッドやプロセスでタスクが実行されるため、エラーが発生してもメインスレッドに直接伝播しない可能性があるからです。

  • エラーハンドリング: result()メソッドやexception()メソッドを呼び出す際に、try...exceptブロックで囲んで例外をキャッチし、適切に処理する必要があります。
  • タイムアウト設定: result(timeout)メソッドを使用すると、指定された時間内に結果が利用可能にならない場合にTimeoutError例外が発生します。これにより、処理がいつまでも終わらないタスクを強制的に中断することができます。

実践的なテクニック

  • as_completed()関数を使うことで、タスクが完了した順に結果を処理できるため、全体的な処理時間を短縮できます。特に、Webスクレイピングのように、レスポンス時間が異なる複数のWebサイトからデータを収集する場合に有効です。
  • コールバック関数を使うことで、タスクの完了時に特定の処理を自動的に実行できます。例えば、タスクの結果をデータベースに書き込んだり、ログを出力したりする場合に便利です。

Futureオブジェクトを使いこなすことで、非同期処理の結果を安全かつ効率的に取得し、より堅牢な並行処理プログラムを開発することができます。エラーハンドリングとタイムアウト設定を適切に行い、as_completed()関数やコールバック関数を積極的に活用することで、並行処理のメリットを最大限に引き出しましょう。

実践例：Webスクレイピングを並行処理で高速化

Webスクレイピングは、ウェブページから情報を抽出する便利な技術ですが、複数のページを処理する場合、時間がかかることがあります。そこで、concurrent.futuresを活用することで、Webスクレイピングを劇的に高速化できます。ここでは、具体的なコード例を通して、並行処理のメリットを体感してみましょう。

Webスクレイピングにおける並行処理

Webスクレイピングは、ネットワークI/Oに時間がかかる処理です。つまり、ウェブサーバーからデータをダウンロードする間、プログラムは待機状態になります。このようなI/Oバウンドな処理は、並行処理と非常に相性が良いです。concurrent.futuresのThreadPoolExecutorを使うことで、複数のウェブページからのデータ取得を並行して行うことができます。

実装例：複数のウェブサイトからタイトルを取得する

以下のコードは、複数のウェブサイトからタイトルを取得する簡単な例です。

“`python
# Webスクレイピングのサンプルコードを実行する前に、以下を実行してBeautiful Soupをインストールしてください
# pip install beautifulsoup4

import concurrent.futures
import requests
from bs4 import BeautifulSoup
import time

def fetch_title(url):
try:
response = requests.get(url, timeout=5)
response.raise_for_status() # HTTPエラーをチェック
soup = BeautifulSoup(response.text, ‘html.parser’)
return url, soup.title.text.strip() if soup.title else ‘タイトルなし’
except requests.exceptions.RequestException as e:
return url, str(e)

urls = [
‘https://www.example.com’,
‘https://www.python.org’,
‘https://www.wikipedia.org’,
‘https://www.google.com’,
‘https://www.yahoo.com’
]

start_time = time.time()

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
future_to_url = {executor.submit(fetch_title, url): url for url in urls}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
url, data = future.result()
print(f'{url}: {data}’)
except Exception as exc:
print(f'{url} generated an exception: {exc}’)

end_time = time.time()
print(f”処理時間: {end_time – start_time:.2f}秒”)
“`

このコードでは、ThreadPoolExecutorを使って、fetch_title関数を複数のスレッドで並行して実行しています。as_completedを使うことで、タスクが完了した順に結果を取得し、処理時間を短縮しています。

コードの解説

  1. `fetch_title(url)`関数: 指定されたURLからウェブページをダウンロードし、Beautiful Soupで解析してタイトルを取得します。エラーが発生した場合は、エラーメッセージを返します。
  2. `urls`リスト: スクレイピング対象のURLのリストです。
  3. `ThreadPoolExecutor`: max_workersで指定された数のスレッドを生成し、タスクを並行して実行します。
  4. `executor.submit(fetch_title, url)`: fetch_title関数を非同期に実行し、Futureオブジェクトを返します。
  5. `concurrent.futures.as_completed(future_to_url)`: 完了したFutureオブジェクトを順番に返します。
  6. `future.result()`: Futureオブジェクトの結果を取得します。エラーが発生した場合は、例外が発生します。

並行処理のメリット

この例では、たった5つのURLですが、URLの数が多ければ多いほど、並行処理の効果が大きくなります。シングルスレッドで順番に処理する場合と比較して、大幅な時間短縮が期待できます。

注意点

  • Webサイトへの負荷: 短時間に大量のリクエストを送信すると、Webサイトに負荷をかけてしまう可能性があります。time.sleep()などで適切な間隔を設けるようにしましょう。
  • robots.txt: スクレイピングを行う前に、robots.txtを確認し、アクセスが許可されているか確認しましょう。
  • 利用規約: Webサイトの利用規約を遵守し、許可されていない方法でデータを取得しないようにしましょう。

まとめ

concurrent.futuresを使うことで、Webスクレイピングを簡単に高速化できます。ぜひ、あなたのスクレイピング処理に取り入れて、効率的なデータ収集を実現してください。

コメント

タイトルとURLをコピーしました