いずみこんな悩みを解決できる記事を書きました！
僕は現役フリーランスエンジニア（歴10年）で、資格は13個保有しています。
「Pythonのasyncioで非同期処理を実装する方法を知りたい」とお考えではありませんか？
複数のAPIリクエストやファイル取得を並行して処理し、待ち時間を減らしたい方へ。
asyncioを使えば、スレッドを増やさずに多数のI/O処理を同時進行できます。
いずみ仕組みを押さえれば書き方はシンプルです。
ということで、本記事ではPythonのasyncioで非同期処理を実装する方法を解説します。
いずみすぐ読み終わるので、ぜひ最後まで読んでくださいませ。
| 【当サイト】おすすめフリーランスエージェント3選 | |||
|---|---|---|---|
| エージェント | 評価 | ポイント | 公式サイト |
 レバテックフリーランス  | 5.0 | 業界最大級のエージェント。 高単価案件が豊富。 | 公式 |
 Midworks  | 4.8 | 満足度調査で 3年連続3冠を達成。 | 公式 |
 ITプロパートナーズ  | 4.6 | 週2〜3向けの案件が豊富。 | 公式 |
【Python】asyncioで非同期処理を実装する方法
早速ですが、Pythonのasyncioで非同期処理を実装する方法を解説します。
asyncioとは何か
asyncioは、Python標準ライブラリに含まれる非同期I/O用のフレームワークです。
1つのスレッド上でイベントループを回し、待ち時間が発生する処理を切り替えながら進めます。
ネットワーク通信やファイルアクセスのように待ち時間が多い処理では、asyncioで全体の実行時間を大きく短縮できます。
一方でCPUを使い続ける計算処理には向かず、計算主体の並列化にはmultiprocessingを使います。
いずみざっくり言うと「待っている間に別の仕事を進める」仕組みですね。
基本の書き方（async / await / asyncio.run）
まずは最小構成のコルーチンを動かします。
async def で定義した関数はコルーチンになります。
コルーチン内では await で他のコルーチンの完了を待てます。
asyncio.run にコルーチンを渡すと、イベントループの作成から後始末までを一括で行います。
import asyncio
async def say_hello():
print("開始")
await asyncio.sleep(1)
print("完了")
asyncio.run(say_hello())await asyncio.sleep(1) の待機中、イベントループは他の処理へ制御を渡せます。
いずみtime.sleepではなくasyncio.sleepを使う点が重要です。
複数の処理を並行実行する（asyncio.gather）
複数のコルーチンを同時に走らせるには asyncio.gather を使います。
import asyncio
import time
async def fetch(name, delay):
await asyncio.sleep(delay)
return f"{name} 完了"
async def main():
start = time.time()
results = await asyncio.gather(
fetch("A", 2),
fetch("B", 1),
fetch("C", 3),
)
print(results)
print(f"経過: {time.time() - start:.1f}秒")
asyncio.run(main())3つのfetchは合計6秒ぶんの待ち時間がありますが、並行実行によって約3秒で終わります。
gather に渡した順番どおりに、戻り値がリストで返ります。
いずみ直列に await すると6秒かかる処理が、gatherなら最も遅いタスクの時間で済みます。
asyncio.TaskGroupで安全に並行実行する
Python 3.11以降では asyncio.TaskGroup が追加され、より安全な並行実行が可能になりました。
import asyncio
async def worker(name, delay):
await asyncio.sleep(delay)
print(f"{name} 完了")
async def main():
async with asyncio.TaskGroup() as tg:
tg.create_task(worker("A", 2))
tg.create_task(worker("B", 1))
tg.create_task(worker("C", 3))
print("すべてのタスクが完了")
asyncio.run(main())async with のブロックを抜けるとき、TaskGroup は登録済みのタスクすべての完了を待ちます。
いずれかのタスクが例外を投げると、TaskGroup は残りのタスクを自動でキャンセルします。
gather はデフォルトでは他タスクをキャンセルしないため、後片付けの確実さで TaskGroup が有利です。
複数タスクが同時に失敗した場合、送出される例外は ExceptionGroup にまとめられます。
ExceptionGroup は except* 構文で種類ごとに受け取れます。
import asyncio
async def ng():
raise ValueError("失敗")
async def ok():
await asyncio.sleep(1)
print("okは完了")
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(ng())
tg.create_task(ok())
except* ValueError as eg:
print(f"捕捉した例外: {eg.exceptions}")
asyncio.run(main())新しいコードで並行実行を書くなら、TaskGroupを第一候補にしています。
タイムアウトを設定する（asyncio.timeout）
応答が返らない処理を打ち切るには asyncio.timeout を使います。
import asyncio
async def slow():
await asyncio.sleep(10)
async def main():
try:
async with asyncio.timeout(3):
await slow()
except TimeoutError:
print("3秒でタイムアウト")
asyncio.run(main())asyncio.timeout も Python 3.11で追加された非同期コンテキストマネージャです。
指定秒数を超えると内側の処理がキャンセルされ、TimeoutError が送出されます。
個別のコルーチン単位で待ち時間を制限したい場合は asyncio.wait_for も使えます。
同期処理でイベントループを止めない（asyncio.to_thread）
asyncioと相性が悪いのが、awaitできない同期的なブロッキング処理です。
requests など同期ライブラリの呼び出しをそのまま書くと、イベントループ全体が止まります。
同期関数を別スレッドへ逃がすには asyncio.to_thread を使います。
import asyncio
def blocking_io():
with open("large.log", "rb") as f:
return f.read()
async def main():
data = await asyncio.to_thread(blocking_io)
print(len(data))
asyncio.run(main())blocking_io の実行中も、イベントループは他のコルーチンを進められます。
いずみHTTP通信なら、はじめから非同期対応の httpx や aiohttp を使う手もあります。
| 【当サイト】おすすめフリーランスエージェント3選 | |||
|---|---|---|---|
| エージェント | 評価 | ポイント | 公式サイト |
レバテックフリーランス | 5.0 | 業界最大級のエージェント。 高単価案件が豊富。 | 公式 |
Midworks | 4.8 | 満足度調査で 3年連続3冠を達成。 | 公式 |
ITプロパートナーズ | 4.6 | 週2〜3向けの案件が豊富。 | 公式 |
【厳選】フリーランスエンジニアにおすすめなエージェント3選
フリーランスエンジニアになるにはエージェントから案件をもらう必要があります。
僕が実際に利用しているおすすめエージェントを紹介しますね。
いずみエージェントは必ず複数登録してください。
担当者によっては「全然案件紹介してくれない…」みたいなこともあるので…
僕は実際に5つのエージェントを使い回していますよ。
フリーランスを始めるなら「レバテックフリーランス
」
| 案件数 | マージン率 | 単価 | |
| 約88,000件 | 非公開 | 高 | |
| 初心者 | 福利厚生 | 申し込み | |
| 無料 |
- 業界最大級の案件数。
- 業界トップクラスの高単価報酬、低マージン（平均年収862万円）。
- 案件参画中のフォローの充実。
※詳細は「【業界最大手】レバテックフリーランスとは？メリットや利用手順を解説！」を参照。
レバテックフリーランス
は業界最大手のフリーランスエージェントです。
とにかく案件数が多いので、とりあえず登録しておけば間違いないエージェントです！
いずみ僕もはじめてフリーランスの案件を貰ったのはレバテックフリーランス
です。
保有している案件数が多いので、業務経験がなくても何かしらの案件は紹介してもらえますよ（僕はJavaの経験3年でも案件を貰えました）。
手厚い保障を重視したいなら「Midworks
」
| 案件数 | マージン率 | 単価 | |
| 約10,000件 | 非公開 | 中 | |
| 初心者 | 福利厚生 | 申し込み | |
| 無料 |
- 手厚い保障で正社員並みの安心感。
- 還元率60％超え＆単価公開でクリアな契約。
- 給与保障制度（審査あり）。
Midworks
は手厚い保障が特徴のフリーランスエージェントです。
フリーランスを目指しているけど不安な方や保障を重視したい方におすすめです。
いずみ僕も何度か案件を紹介してもらいました。
自分のスキルに合った案件を紹介してもらえましたし、電話のやり取りも非常に丁寧でした。
週2〜3日の案件探しなら「ITプロパートナーズ
」
| 案件数 | マージン率 | 単価 | |
| 約5,000件 | 非公開 | 高 | |
| 初心者 | 福利厚生 | 申し込み | |
| 経験者向け | 無料 |
- IT案件に特化したフリーランスエージェント。
- 週2〜3日の案件が豊富。
- リモート案件が多く、直エンドなので単価も高い。
※詳細は「【週2・3案件】ITプロパートナーズとは？メリットや利用手順を解説！」を参照。
ITプロパートナーズ
はIT案件に特化したフリーランスエージェントです。
週2〜3日から参画できる案件が豊富なので、起業したい人にもおすすめです。
いずみ週2〜3日の案件はある程度スキルがないと紹介してもらえない印象です。
とはいえ、週5の案件ももちろんありますし、僕が利用した時は迅速・丁寧に対応していただきました！
よくある質問
asyncioとマルチスレッド（threading）はどう使い分けますか？
I/O待ちが中心で同時接続数が多い場面では asyncio が向いています。
ライブラリが同期APIしか提供していない場合や、少数の並行処理で十分な場合は threading が手軽です。
gatherとTaskGroupはどちらを使うべきですか？
Python 3.11以降を使えるなら、エラー時の挙動が明確な TaskGroup を推奨します。
3.10以前をサポートする必要がある場合は gather を使い、return_exceptions 引数で例外の扱いを調整します。
コルーチン内でtime.sleepを使うとどうなりますか？
time.sleep はイベントループごと停止させるため、並行実行の効果が失われます。
非同期の待機には必ず asyncio.sleep を使います。
まとめ
今回は、Pythonのasyncioで非同期処理を実装する方法について解説しました。
以下が本記事のまとめになります。
- おすすめ本
Pythonの勉強なら「独習Python」が体系的に学べるのでおすすめですよ♪
いずみ最後までお読みいただき、ありがとうございました！
| 【当サイト】おすすめフリーランスエージェント3選 | |||
|---|---|---|---|
| エージェント | 評価 | ポイント | 公式サイト |
レバテックフリーランス | 5.0 | 業界最大級のエージェント。 高単価案件が豊富。 | 公式 |
Midworks | 4.8 | 満足度調査で 3年連続3冠を達成。 | 公式 |
ITプロパートナーズ | 4.6 | 週2〜3向けの案件が豊富。 | 公式 |
- クソおすすめ本
いずみ海外のエンジニアがどういった思考で働いているかが理解できます。
海外に行く気はないけど海外エンジニアの動向が気になる雑魚エンジニアにおすすめです（本当におすすめな本しか紹介しないのでご安心を）。
