asyncio or threading? Decide with 3 key questions
1. Overview
Many people learning Python concurrency often ask “is asyncio or threading better?” — this question is flawed from the start, because the two tools solve different problems. This article presents 3 specific questions for you to answer yourself, along with executable code examples. After reading, you’ll know exactly when to use threading, when to use asyncio, and when both are useless (you must use multiprocessing).
2. Why the question “which one is faster” is wrong
Python has a Global Interpreter Lock (GIL) — only one Python thread executes bytecode at any given moment, regardless of how many CPU cores the machine has. Both threading and asyncio cannot break the GIL. Both only prevent the CPU from being “blocked” while waiting for I/O (reading files, network calls, database queries) — they don’t make computation faster.
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"Sequential: {time.perf_counter() - start:.2f}s")
Sequential: 1.85s
Whether you run cpu_bound_task() twice using threading.Thread or asyncio.gather, the total time doesn’t decrease, because the GIL only allows one thread to execute Python code at any moment. This is why the first question below is the most important.
Note: If your problem is CPU-bound (heavy arithmetic, image processing, encryption…), both
threadingandasynciowon’t help. You needmultiprocessingor write the heavy part in C/Rust extensions (NumPy, Cython).
3. Question 1: Is your work I/O-bound or CPU-bound?
This is the question that determines whether you should use “waiting”-type concurrency (I/O-bound) or not.
3.1. I/O-bound: both threading and asyncio are efficient
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 runs 5 requests concurrently using 5 OS threads. Because requests.get is a blocking call waiting for network, the GIL is released when the thread is waiting for I/O at the C level — so 5 requests run truly in parallel, total time ~1s instead of 5s.
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]
Results are equivalent to threading, but the mechanism differs: only one thread, the event loop actively awaits and yields control when waiting for I/O, no need for the OS to create new threads.
3.2. CPU-bound: both lose to 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 creates separate processes, each with its own GIL, so it can truly utilize multiple CPU cores. If you’re wavering between “threading or asyncio” for a CPU-bound problem, the correct answer is both are wrong — use multiprocessing.
4. Question 2: Does your ecosystem support async?
This is the most practical question when working in a real codebase, where you don’t write everything from scratch.
import psycopg2 # synchronous driver, blocking
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 is a blocking library — it has no awaitable version. To run multiple queries in parallel with this driver, threading (or asyncio’s run_in_executor) is the natural choice, because you don’t need to rewrite anything.
import asyncpg # async-native driver
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"]}
If your ecosystem already has async-native drivers (asyncpg, aiohttp, motor for MongoDB…), asyncio maximizes its advantage: thousands of concurrent connections on one thread, without the overhead of creating OS threads.
Note: Mixing blocking code into
async defwithout properawaiting (e.g., callingrequests.getdirectly in a coroutine) will block the entire event loop — all other tasks must wait, even unrelated ones. This is the most common mistake when switching from threading to asyncio.
5. Question 3: How many concurrent tasks do you need?
The resource overhead between threads and coroutines differs greatly as the number of tasks increases.
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"Peak RSS with 2000 threads: {peak_kb / 1024:.1f} MB")
for t in threads:
t.join()
Peak RSS with 2000 threads: 1850.3 MB
Each OS thread by default takes several hundred KB to several MB of stack, so 2000 threads easily consume nearly 2 GB of RAM and create heavy context-switching pressure on the OS scheduler.
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"Peak RSS with 2000 coroutines: {peak_kb / 1024:.1f} MB")
asyncio.run(main())
Peak RSS with 2000 coroutines: 95.7 MB
Coroutines are pure Python objects, no separate OS thread needed — the overhead of creating 2000 tasks is only tens of MB. If you need to handle thousands of concurrent connections (web server, chat server, large-scale crawler), asyncio is nearly mandatory; threading will run out of resources before bandwidth.
6. Quick decision table
| Question | Answer → Choice |
|---|---|
| 1. I/O-bound or CPU-bound? | CPU-bound → multiprocessing. I/O-bound → proceed to question 2 |
| 2. Does your library have async-native support? | No (old blocking driver) → threading. Yes (aiohttp, asyncpg…) → asyncio |
| 3. How many concurrent tasks do you need? | Tens → threading is simple enough. Hundreds/thousands → asyncio |
In practice, many systems use both: asyncio for large-scale network I/O-bound layer, combined with run_in_executor to run blocking code segments (e.g., old libraries) in a separate thread pool to avoid blocking the event loop.
import asyncio
def legacy_blocking_call(x: int) -> int:
import time
time.sleep(1) # simulate old blocking library
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 pushes the blocking function to a thread in the default thread pool, the event loop remains free to handle other coroutines while waiting.
7. Testing behavior with pytest
Example test confirming asyncio.gather runs tasks in true parallel, not sequentially:
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]
# If run sequentially would take ~1.5s, running in parallel only ~0.5s
assert elapsed < 0.7
Required dependencies for testing:
pytest
pytest-asyncio
This test confirms the true nature of asyncio: three sleep(0.5) tasks running concurrently on one thread cost only ~0.5s total, not 1.5s — if the elapsed < 0.7 assertion fails, it means your code is awaiting sequentially instead of using gather.
8. Common mistakes
- Using
time.sleep()in a coroutine: must useawait asyncio.sleep().time.sleep()is a blocking call that stops the entire event loop. - Forgetting
await: callingfetch(session, url)withoutawaitonly creates a coroutine object, nothing runs — Python will warnRuntimeWarning: coroutine was never awaited. - Sharing state between threads without locking:
threadinghas real race conditions (multiple threads writing the same variable), needsthreading.Lock.asyncioon a single thread doesn’t have this type of race condition, but can still have logic bugs ifawaithappens in the middle of code assumed to be “atomic”. - Using
ThreadPoolExecutorfor CPU-bound problems then disappointed it’s not faster — because the GIL still blocks, useProcessPoolExecutor.
Note: If your team is comfortable with
threadingand the codebase is full of blocking libraries, don’t rush to refactor toasynciojust because it’s “newer”. The cost of rewriting all drivers, handling exceptions differently, and debugging complex coroutine stack traces is much higher than the benefit you’d get if the number of concurrent tasks isn’t large.
9. Conclusion
Choosing between asyncio and threading isn’t about “which one is faster” but three specific questions: whether your problem is I/O-bound or CPU-bound, whether your library ecosystem supports async-native, and what scale of concurrency you need to handle. CPU-bound always needs multiprocessing. I/O-bound with old libraries and small task counts means threading is simple and easy to integrate. I/O-bound at large scale with async-native libraries means asyncio is the right choice for performance and resources.