はじめに

こんにちは！今回は、Pythonにおける並行処理と並列処理について、マルチスレッド、マルチプロセス、非同期プログラミングの3つのアプローチを比較しながら詳しく解説します。これらの技術を理解し、適切に使用することで、効率的で高性能なPythonプログラムを作成することができます。

1. 並行処理と並列処理の違い

まず、並行処理（Concurrency）と並列処理（Parallelism）の違いを理解することが重要です。

並行処理: 複数のタスクを切り替えながら実行すること。同時に進行しているように見えるが、実際には1つの処理単位で実行している。
並列処理: 複数のタスクを同時に実行すること。複数の処理単位（CPUコアなど）で実際に同時に処理を行う。

Pythonでは、これらを実現するために主に以下の3つのアプローチがあります：

マルチスレッド
マルチプロセス
非同期プログラミング

それぞれについて詳しく見ていきましょう。

2. マルチスレッド

マルチスレッドは、1つのプロセス内で複数の実行スレッドを使用する方法です。

2.1 基本的な使い方

import threading
import time

def worker(name):
    print(f"Worker {name} starting")
    time.sleep(2)
    print(f"Worker {name} finished")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("All workers finished")

2.2 特徴

メモリを共有するため、データの共有が容易
I/O束縛のタスクに適している
CPUバウンドのタスクには適していない（GILの制限により）
デッドロックなどの同期の問題に注意が必要

2.3 Global Interpreter Lock (GIL)

Pythonのマルチスレッドを理解する上で、GILは重要な概念です。GILは、Pythonインタープリタが同時に1つのスレッドしか実行できないようにする仕組みです。これにより、CPUバウンドのタスクではマルチスレッドによる性能向上が期待できません。

3. マルチプロセス

マルチプロセスは、複数のPythonプロセスを並行して実行する方法です。

3.1 基本的な使い方

import multiprocessing
import time

def worker(name):
    print(f"Worker {name} starting")
    time.sleep(2)
    print(f"Worker {name} finished")

if __name__ == '__main__':
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("All workers finished")

3.2 特徴

別々のメモリ空間で実行されるため、GILの影響を受けない
CPUバウンドのタスクに適している
プロセス間通信にはオーバーヘッドがある
メモリ使用量が増加する

3.3 プロセス間通信

マルチプロセスでデータを共有するには、特別な方法が必要です。例えば、multiprocessing.Queueを使用できます：

from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # [42, None, 'hello']を出力
    p.join()

4. 非同期プログラミング

非同期プログラミングは、コルーチンを使用して並行処理を実現する方法です。Python 3.5以降では、async/await構文を使用できます。

4.1 基本的な使い方

import asyncio

async def worker(name):
    print(f"Worker {name} starting")
    await asyncio.sleep(2)
    print(f"Worker {name} finished")

async def main():
    tasks = [asyncio.create_task(worker(i)) for i in range(5)]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    asyncio.run(main())

4.2 特徴

1つのスレッドで動作するため、同期の問題が少ない
I/O束縛のタスクに非常に適している
CPUバウンドのタスクには適していない
コードの構造が変わるため、既存のコードの修正が必要

4.3 イベントループ

非同期プログラミングの核心は、イベントループです。イベントループは、タスクのスケジューリングと実行を管理します。

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

loop = asyncio.get_event_loop()
loop.run_until_complete(hello())
loop.close()

5. パフォーマンス比較

それぞれのアプローチのパフォーマンスを比較するために、簡単なベンチマークを行ってみましょう。ここでは、I/O束縛のタスクとCPU束縛のタスクの両方をテストします。

import time
import threading
import multiprocessing
import asyncio

def io_bound(duration):
    time.sleep(duration)

def cpu_bound(n):
    return sum(i * i for i in range(n))

# マルチスレッド
def run_threads(func, args, n):
    threads = [threading.Thread(target=func, args=args) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

# マルチプロセス
def run_processes(func, args, n):
    processes = [multiprocessing.Process(target=func, args=args) for _ in range(n)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

# 非同期
async def async_io_bound(duration):
    await asyncio.sleep(duration)

async def run_async(func, args, n):
    await asyncio.gather(*[func(*args) for _ in range(n)])

def benchmark(name, func, args, n):
    start = time.time()
    func(*args, n)
    end = time.time()
    print(f"{name}: {end - start:.2f} seconds")

if __name__ == '__main__':
    print("I/O Bound Task")
    benchmark("Sequential", lambda d, n: [io_bound(d) for _ in range(n)], (1,), 5)
    benchmark("Threading", run_threads, (io_bound, (1,)), 5)
    benchmark("Multiprocessing", run_processes, (io_bound, (1,)), 5)
    benchmark("Asyncio", asyncio.run, (run_async(async_io_bound, (1,), 5),), 1)

    print("\nCPU Bound Task")
    benchmark("Sequential", lambda n, count: [cpu_bound(n) for _ in range(count)], (10**6,), 5)
    benchmark("Threading", run_threads, (cpu_bound, (10**6,)), 5)
    benchmark("Multiprocessing", run_processes, (cpu_bound, (10**6,)), 5)

出力例：

I/O Bound Task
Sequential: 5.01 seconds
Threading: 1.00 seconds
Multiprocessing: 1.02 seconds
Asyncio: 1.00 seconds

CPU Bound Task
Sequential: 2.34 seconds
Threading: 2.36 seconds
Multiprocessing: 0.62 seconds

この結果から、以下のことが分かります：

I/O束縛のタスクでは、マルチスレッド、マルチプロセス、非同期プログラミングのいずれも効果的です。
CPU束縛のタスクでは、マルチプロセスが最も効果的です。マルチスレッドはGILの影響で逐次実行とほぼ同じ性能です。

6. 使い分けの指針

マルチスレッド
- I/O束縛のタスクで、共有メモリが必要な場合
- 例：複数のネットワークリクエストを並行して処理する場合
マルチプロセス
- CPUバウンドのタスクで、並列処理が必要な場合
- 例：大規模な数値計算や画像処理
非同期プログラミング
- 多数のI/O操作を効率的に処理する必要がある場合
- 例：Web スクレイピング、大量のAPIリクエスト

まとめ

Pythonの並行処理と並列処理には、マルチスレッド、マルチプロセス、非同期プログラミングという3つの主要なアプローチがあります。それぞれに長所と短所があり、タスクの性質に応じて適切な方法を選択することが重要です。

マルチスレッドは、I/O束縛のタスクに適していますが、GILの影響でCPUバウンドのタスクには向いていません。
マルチプロセスは、CPUバウンドのタスクに最適ですが、メモリ使用量が増加し、プロセス間通信にオーバーヘッドがあります。
非同期プログラミングは、多数のI/O操作を効率的に処理できますが、既存のコードの大幅な書き換えが必要になる場合があります。

適切なアプローチを選択し、効率的に実装することで、Pythonプログラムのパフォーマンスを大幅に向上させることができます。状況に応じて最適な方法を選び、並行処理と並列処理の力を最大限に活用しましょう。

以上、Pythonの並行処理と並列処理についての記事でした。ご清読ありがとうございました！

関係する記事