Python並行処理入門：効率的なコード開発への道【初心者向け】

www.amazon.co.jp

Python並行処理入門：効率的なコード開発への道【初心者向け】

Pythonにおける並行処理は、プログラムのパフォーマンスと応答性を向上させるための強力なツールです。特にI/Oバウンドな処理（ネットワーク通信やファイル操作など）が多いアプリケーションでは、並行処理を導入することで大幅な改善が見込めます。本記事では、Pythonの並行処理の基礎から応用までを段階的に解説し、20個の練習問題を通して実践的なスキルを習得できるようサポートします。

はじめに

並行処理は、複数のタスクを同時に実行する技術全般を指します。しかし、Pythonの標準インタープリタ（CPython）では、真の意味での並列処理（複数のCPUコアを同時に利用する処理）が制限されています。そのため、本記事では主に「擬似並行処理」に焦点を当てます。

What is Concurrency? Concurrency refers to the ability of a program to handle multiple tasks seemingly simultaneously. In Python, due to the Global Interpreter Lock (GIL), true parallelism is limited, so we primarily focus on concurrency techniques that allow for efficient handling of I/O-bound operations.

擬似並行処理は、単一のCPUコア上で複数のタスクを時間分割して実行することで、あたかも同時に実行されているかのように見せる手法です。これにより、I/O待ちが多いプログラムでは、他のタスクに処理を譲ることで、全体的な処理効率を向上させることができます。

Why is Concurrency Important? Concurrency becomes crucial when dealing with I/O-bound tasks, such as network requests or file operations. By allowing other tasks to run while one task waits for I/O, you can significantly improve the overall responsiveness and throughput of your application.

Pythonには、並行処理を実現するためのいくつかの主要な手法があります。

• threading: スレッド（軽量なプロセス）を利用した並行処理
• multiprocessing: プロセスを利用した並行処理
• asyncio: 非同期I/Oモデルを用いた並行処理

それぞれの特徴を理解し、適切な手法を選択することが重要です。

threading：スレッドによる並行処理

threadingは、軽量なプロセスであるスレッドを利用して並行処理を実現する方法です。スレッドは同じメモリ空間を共有するため、プロセス間通信のオーバーヘッドが少なく、高速に動作します。しかし、PythonのCPythonインタープリタにはGIL（グローバルインタープリタロック）が存在し、一度に1つのスレッドしかPythonバイトコードを実行できません。そのため、CPUバウンドな処理では並列化の効果は限定的ですが、I/O待ちが多い処理やGUIアプリケーションなどでは有効です。

What is Threading? Threading involves creating multiple threads within a single process. Threads share the same memory space, allowing for efficient communication but also introducing potential concurrency issues like race conditions if not managed carefully. The Global Interpreter Lock (GIL) in CPython limits true parallelism with threading, making it more suitable for I/O-bound tasks.

threadingのメリット:

• プロセス間通信のオーバーヘッドが少ない
• GUIアプリケーションとの連携が容易

threadingのデメリット:

• GILの影響を受けるため、CPUバウンドな処理には不向き
• スレッドセーフなコードを記述する必要がある（競合状態を防ぐためにロックなどの同期機構を使用）

threadingの練習問題

1. スレッドの作成と実行: threading.Threadクラスを使って、"Hello from thread!"というメッセージを5秒ごとに表示するスレッドを作成し、実行してください。

import threading
import time

def print_hello():
    while True:
        print("Hello from thread!")
        time.sleep(5)

thread = threading.Thread(target=print_hello)
thread.start()

2. スレッドの終了: 上記の問題で作成したスレッドを、一定時間後に安全に終了させるように修正してください。threading.Eventクラスを利用すると便利です。

import threading
import time

def print_hello(event):
    while not event.is_set():
        print("Hello from thread!")
        time.sleep(5)

event = threading.Event()
thread = threading.Thread(target=print_hello, args=(event,))
thread.start()

time.sleep(10)  # 10秒後にスレッドを終了させる
event.set()
thread.join() # スレッドが終了するまで待機

3. スレッドセーフなカウンタ: 複数のスレッドから同時にアクセスされるカウンタ変数を定義し、各スレッドで10回ずつインクリメントするようにしてください。threading.Lockクラスを使って、競合状態を防ぎましょう。

import threading

counter = 0
lock = threading.Lock()

def increment_counter():
    global counter
    for _ in range(10):
        with lock:  # ロックを獲得してカウンタをインクリメント
            counter += 1

threads = []
for _ in range(5):
    thread = threading.Thread(target=increment_counter)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("Final counter value:", counter)  # 最終的なカウンタの値は50になるはず

4. スレッド間通信: threading.Queueクラスを使って、複数のスレッド間でメッセージをやり取りするプログラムを作成してください。あるスレッドがメッセージをキューに投入し、別のスレッドがキューからメッセージを取り出して表示するようにします。

import threading
import time
import queue

q = queue.Queue()

def producer():
    for i in range(5):
        message = f"Message {i}"
        q.put(message)
        print(f"Producer: Put '{message}' into the queue")
        time.sleep(1)

def consumer():
    while True:
        try:
            message = q.get(timeout=2)  # タイムアウトを設定
            print(f"Consumer: Got '{message}' from the queue")
            q.task_done() # キュー内のタスク完了を通知
        except queue.Empty:
            print("Queue is empty, exiting consumer.")
            break

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
q.join() # キュー内のすべてのタスクが完了するまで待機
consumer_thread.join()

5. スレッドプール: concurrent.futures.ThreadPoolExecutorを使って、複数のスレッドを効率的に管理し、タスクを分散実行するプログラムを作成してください。10個の数値を受け取り、それぞれの数値の平方根を計算するタスクをスレッドプールに投入します。

import concurrent.futures
import math

def calculate_square_root(number):
    return math.sqrt(number)

numbers = [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(calculate_square_root, numbers))

print("Square roots:", results)

multiprocessing：プロセスによる並行処理

multiprocessingは、複数のプロセスを利用して並行処理を実現する方法です。各プロセスは独自のメモリ空間を持つため、GILの影響を受けず、CPUバウンドな処理でも並列化の効果を発揮します。ただし、プロセス間通信のオーバーヘッドがあるため、I/O待ちが多い処理ではthreadingの方が効率的な場合があります。

What is Multiprocessing? Multiprocessing involves creating multiple processes, each with its own memory space. This allows for true parallelism on multi-core systems and bypasses the limitations of the Global Interpreter Lock (GIL). However, inter-process communication can be more complex and introduce overhead compared to threading.

multiprocessingのメリット:

• GILの影響を受けないため、CPUバウンドな処理に有効
• 複数のコアを有効活用できる

multiprocessingのデメリット:

• プロセス間通信のオーバーヘッドがある
• メモリ消費量が多い

multiprocessingの練習問題

6. プロセスの作成と実行: multiprocessing.Processクラスを使って、"Hello from process!"というメッセージを5秒ごとに表示するプロセスを作成し、実行してください。

import multiprocessing
import time

def print_hello():
    while True:
        print("Hello from process!")
        time.sleep(5)

process = multiprocessing.Process(target=print_hello)
process.start()

7. プロセスの終了: 上記の問題で作成したプロセスを、一定時間後に安全に終了させるように修正してください。multiprocessing.Eventクラスを利用すると便利です。

import multiprocessing
import time

def print_hello(event):
    while not event.is_set():
        print("Hello from process!")
        time.sleep(5)

event = multiprocessing.Event()
process = multiprocessing.Process(target=print_hello, args=(event,))
process.start()

time.sleep(10)  # 10秒後にプロセスを終了させる
event.set()
process.join() # プロセスが終了するまで待機

8. プロセス間通信: multiprocessing.Queueクラスを使って、複数のプロセス間でメッセージをやり取りするプログラムを作成してください。あるプロセスがメッセージをキューに投入し、別のプロセスがキューからメッセージを取り出して表示するようにします。

import multiprocessing
import time
import queue

def producer(q):
    for i in range(5):
        message = f"Message {i}"
        q.put(message)
        print(f"Producer: Put '{message}' into the queue")
        time.sleep(1)

def consumer(q):
    while True:
        try:
            message = q.get(timeout=2)  # タイムアウトを設定
            print(f"Consumer: Got '{message}' from the queue")
        except queue.Empty:
            print("Queue is empty, exiting consumer.")
            break

q = multiprocessing.Queue()
producer_process = multiprocessing.Process(target=producer, args=(q,))
consumer_process = multiprocessing.Process(target=consumer, args=(q,))

producer_process.start()
consumer_process.start()

producer_process.join()
q.close() # キューを閉じる
q.clear()  # キューの内容をクリアする
consumer_process.join()

9. プロセスプール: concurrent.futures.ProcessPoolExecutorを使って、複数のプロセスを効率的に管理し、タスクを分散実行するプログラムを作成してください。10個の数値を受け取り、それぞれの数値の階乗を計算するタスクをプロセスプールに投入します。

import concurrent.futures

def calculate_factorial(number):
    if number == 0:
        return 1
    else:
        result = 1
        for i in range(1, number + 1):
            result *= i
        return result

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(calculate_factorial, numbers))

print("Factorials:", results)

asyncio：非同期I/Oによる効率化

asyncioは、Pythonで非同期I/Oプログラミングを容易にするためのライブラリです。イベントループを使って、複数のコルーチンを同時に実行し、I/O待ち時間を効率的に利用することができます。特に、ネットワークアプリケーションやWebサーバーなど、I/O処理が多いプログラムでは効果を発揮します。

What is asyncio? asyncio provides a framework for writing concurrent code using the asynchronous I/O (async/await) model. It uses an event loop to manage multiple coroutines, allowing efficient handling of I/O-bound operations without blocking. This is particularly useful in network applications and web servers where waiting for I/O can be a bottleneck.

asyncioのメリット:

• I/O待ち時間を効率的に利用できる
• シングルスレッドで複数のタスクを同時に実行できる
• ネットワークアプリケーションやWebサーバーに適している

asyncioのデメリット:

• イベントループの概念を理解する必要がある
• CPUバウンドな処理には不向き

asyncioの練習問題

10. コルーチンの作成: async defキーワードを使って、簡単なコルーチンを作成し、実行してください。

import asyncio

async def my_coroutine():
    print("Coroutine started")
    await asyncio.sleep(1)  # 1秒間スリープ
    print("Coroutine finished")

asyncio.run(my_coroutine())

11. 複数のコルーチンの同時実行: asyncio.gatherを使って、複数のコルーチンを同時に実行するプログラムを作成してください。

import asyncio

async def my_coroutine(id):
    print(f"Coroutine {id} started")
    await asyncio.sleep(1)  # 1秒間スリープ
    print(f"Coroutine {id} finished")

async def main():
    tasks = [my_coroutine(1), my_coroutine(2), my_coroutine(3)]
    await asyncio.gather(*tasks)

asyncio.run(main())

12. 非同期I/O: aiohttpライブラリを使って、複数のURLから同時にデータを取得するプログラムを作成してください。

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

async def main():
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://docs.python.org/3/"
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"Content from {urls[i]}: {result[:100]}...")  # 最初の100文字を表示

asyncio.run(main())

13. イベントループ: asyncio.create_taskを使って、複数のタスクをイベントループに登録し、同時に実行するプログラムを作成してください。

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    task1_coroutine = task1()
    task2_coroutine = task2()

    task1 = asyncio.create_task(task1_coroutine)
    task2 = asyncio.create_task(task2_coroutine)

    await asyncio.gather(task1, task2)

asyncio.run(main())

14-20: 上記で学んだ知識を応用して、より複雑な並行処理のシナリオに挑戦してみてください。例えば、以下のような課題に取り組むことができます。

• 複数のスレッド/プロセスを使って、ファイルから大量のデータを読み込み、処理するプログラムを作成する。
• asyncioを使って、WebSocketサーバーを実装し、クライアントからのメッセージを非同期的に処理する。
• 並行処理のパフォーマンスを計測し、最適な並行度を見つける。
• デッドロックや競合状態が発生しやすい状況を作り出し、その解決策を検討する。

まとめ

今回は、Pythonにおける並行処理の基礎から応用までを、20問の練習問題を通して解説しました。threading, multiprocessing, asyncioそれぞれの特徴を理解し、適切な手法を選択することで、プログラムの効率性と応答性を大幅に向上させることができます。

これらの練習問題を参考に、ぜひ実際にコードを書いて、並行処理の世界を探求してみてください。 より深く学ぶためには、以下のリソースも参照することをお勧めします。

• Python公式ドキュメント: https://docs.python.org/ja/3/library/concurrent.futures.html
• asyncioのドキュメント: https://docs.python.org/ja/3/library/asyncio.html
• Python threading tutorial: https://realpython.com/python-threading-multithreading/

並行処理は奥深いテーマですが、一つずつ丁寧に理解していくことで、必ずマスターできるはずです。頑張ってください！