更快的 Python：async/await 和 threading 中的并发
如果您使用 Python 编程已经有一段时间，特别是使用过 FastAPI 和 discord.py 这样的框架和库，那么您可能已经接触过 async/await 或 asyncio。 您也许听说过“Python 中不存在多线程处理”这种说法，也可能知道 Python 中大名鼎鼎（或臭名昭著）的 GIL。 考虑到 Python 中多线程受到的否定，您可能好奇 async/await 与多线程处理之间到底有什么区别，尤其是在 Python 编程中。 如果上述情况与您相符，那么这篇博文就是为您准备的！
什么是多线程处理？
在编程中，多线程处理是指程序同时执行多个顺序任务（即线程）的能力。 这些线程可以在单个处理器核心上运行，也可以跨多个核心运行。 不过，由于全局解释器锁 (GIL) 的限制，Python 中的多线程处理只能在单个核心上处理。 nogil（也称无线程）Python 是个例外，它消除了 GIL，本系列的第二部分将对此进行介绍。 在这篇博文中，我们假设 GIL 始终存在。
什么是并发？
编程中的并发是指计算机同时执行多个任务，或者似乎在同时执行多个任务，即使不同的任务是在单个处理器上执行。 通过管理程序不同部分之间的资源和交互，不同的任务可以在重叠时间间隔内独立取得进展。
asyncio 和 threading 在 Python 中都表现为并发
粗略地说，asyncio 和 threading Python 库都支持并发。 不过，您的 CPU 并没有在完全相同的时间执行多个任务， 只是看起来像而已。
假设您要为几位客人准备一顿多道菜的晚餐。 有些菜需要时间烹制，例如，烤箱中烘烤的派或者炉子上慢炖的汤。 在这些菜肴烹制期间，我们不会干等着， 而是会去忙其他事。 这就类似于 Python 中的并发。 有时，Python 进程会等待某个任务完成。 例如，当一些输入/输出 (I/O) 进程由操作系统处理时，Python 进程只是在等待。 这时，我们可以使用 async 让另一个 Python 进程在等待期间运行。
区别在于谁说了算
如果 asyncio 和 threading 都表现为并发，它们之间有什么区别？ 主要区别在于谁负责哪个进程在什么时候运行。 对于 async/await，这种方法有时被称为合作并发。 一个协程或 future 将其控制权转交给另一个协程或 future。 另一方面，在 threading 中，操作系统的管理器将控制哪个进程运行。
以会议作为类比，在合作并发的会议中，一个麦克风被传来传去。 拿着麦克风的人可以发言，当他发言完毕，他会将麦克风传递给下一个人。 而在多线程处理的会议中，则是由一个主席决定谁在某段时间内可以发言。
在 Python 中编写并发代码
我们编写一些示例代码来了解 Python 中并发的运作方式。 我们将使用 asyncio 和 threading 创建一个快餐店模拟。
Python 中 async/await 的运作方式
asyncio 软件包在 Python 3.4 中引入，而 async 和 await 关键字在 Python 3.5 中引入。 async/await 之所以成为可能，主要原因之一是协程的使用。 Python 中的协程实际上是重新设计的生成器，能够暂停并传回 main 函数。
现在，想象一下一家汉堡店，只有一名员工在工作。 订单按照先进先出的队列准备，不能执行异步操作：
import time
def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
def main():
for i in range(3):
make_burger(i)
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
这将需要一些时间来完成：
Preparing burger #0... Burger made #0 Preparing burger #1... Burger made #1 Preparing burger #2... Burger made #2 Orders completed in 15.01 seconds.
现在，假设餐厅增加了人手，员工们同时工作：
import asyncio
import time
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def main():
order_queue = []
for i in range(3):
order_queue.append(make_burger(i))
await asyncio.gather(*(order_queue))
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
我们可以看到两者的区别：
Preparing burger #0... Preparing burger #1... Preparing burger #2... Burger made #0 Burger made #1 Burger made #2 Orders completed in 5.00 seconds.
使用 asyncio 提供的函数，例如 run 和 gather，以及关键字 async 和 await，我们创建出可以并发制作汉堡的协程。
接下来，我们更进一步，创建一个更复杂的模拟。 假设我们只有两个工作进程，一次只能制作两个汉堡。
import asyncio
import time
order_queue = asyncio.Queue()
def take_order():
for i in range(3):
order_queue.put_nowait(make_burger(i))
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while order_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await order_queue.get()
await task
print(f"{self.name} finished a task...")
async def main():
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
take_order()
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
我们用队列保存任务，然后由员工领取。
John is working... Preparing burger #0... Jane is working... Preparing burger #1... Burger made #0 John finished a task... John is working... Preparing burger #2... Burger made #1 Jane finished a task... Burger made #2 John finished a task... Orders completed in 10.00 seconds.
在这个示例中，我们使用 asyncio.Queue 存储任务，不过，如果有多种类型的任务，它会更实用，如下例所示。
import asyncio
import time
task_queue = asyncio.Queue()
order_num = 0
async def take_order():
global order_num
order_num += 1
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = input("Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = input("Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
print(f"Order #{order_num:04d} queued.")
await task_queue.put(take_order())
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def make_fries(order_num):
print(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
print(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await task_queue.get()
await task
print(f"{self.name} finish task...")
else:
await asyncio.sleep(1) #rest
async def main():
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
这个示例中有多个任务，包括制作薯条（耗时较少）和接受订单（需要用户输入）。
可以看到程序停止来等待用户输入，甚至其他没有接受订单的员工也停止了后台工作。 这是因为 input 函数不是异步的，因此不会被等待。 记住，异步代码中的控制只在被等待时才会释放。 为了解决这个问题，我们可以将：
input("Number of burgers:")
替换为
await asyncio.to_thread(input, "Number of burgers:")
对薯条也是一样，如以下代码所示。 注意，现在程序在无限循环中运行。 如果需要停止，我们可以故意使用无效输入使程序崩溃。
import asyncio
import time
task_queue = asyncio.Queue()
order_num = 0
async def take_order():
global order_num
order_num += 1
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = await asyncio.to_thread(input, "Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = await asyncio.to_thread(input, "Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
print(f"Order #{order_num:04d} queued.")
await task_queue.put(take_order())
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def make_fries(order_num):
print(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
print(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await task_queue.get()
await task
print(f"{self.name} finish task...")
else:
await asyncio.sleep(1) #rest
async def main():
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
使用 asyncio.to_thread，我们将 input 函数放入了一个单独的线程（参阅此参考）。 不过，需要注意的是，这个技巧只有在 Python GIL 存在时才能疏通 I/O 绑定任务。
如果运行上面的代码，您可能还会发现终端中的标准 I/O 出现混乱。 用户 I/O 和发生的事件的记录应该是分开的。 我们可以将记录放入日志，以后再检查。
import asyncio
import logging
import time
logger = logging.getLogger(__name__)
logging.basicConfig(filename='pyburger.log', level=logging.INFO)
task_queue = asyncio.Queue()
order_num = 0
closing = False
async def take_order():
global order_num, closing
try:
order_num += 1
logger.info(f"Taking Order #{order_num:04d}...")
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = await asyncio.to_thread(input, "Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = await asyncio.to_thread(input, "Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
logger.info(f"Order #{order_num:04d} queued.")
print(f"Order #{order_num:04d} queued, please wait.")
await task_queue.put(take_order())
except ValueError:
print("Goodbye!")
logger.info("Closing down... stop taking orders and finish all tasks.")
closing = True
async def make_burger(order_num):
logger.info(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
logger.info(f"Burger made #{order_num}")
async def make_fries(order_num):
logger.info(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
logger.info(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
logger.info(f"{self.name} is working...")
task = await task_queue.get()
await task
task_queue.task_done()
logger.info(f"{self.name} finish task.")
elif closing:
return
else:
await asyncio.sleep(1) #rest
async def main():
global task_queue
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
print("Welcome to Pyburger!")
logger.info("Ready for business!")
await asyncio.gather(staff1.working(), staff2.working())
logger.info("All tasks finished. Closing now.")
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
logger.info(f"Orders completed in {elapsed:0.2f} seconds.")
在这最后的代码块中，我们将模拟信息记录在 pyburger.log 中，并为客户消息预留了终端。 我们还会在点餐过程中捕获无效输入，并在输入无效时将 closing 标志切换为 True（假设用户想要退出）。 在 closing 标志设为 True 后，工作线程将 return，结束协程的无限 while 循环。
Python 中 threading 的运作方式
在上面的示例中，我们将 I/O 绑定任务放入另一个线程。 您可能想知道，是否可以将所有任务放入单独的线程，并让它们并发运行。 我们尝试使用 threading 而不是 asyncio。
在下面的代码中我们并发制作汉堡，没有限制：
import asyncio
import time
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def main():
order_queue = []
for i in range(3):
order_queue.append(make_burger(i))
await asyncio.gather(*(order_queue))
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
```
Instead of creating async coroutines to make the burgers, we can just send functions down different threads like this:
```
import threading
import time
def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
def main():
order_queue = []
for i in range(3):
task = threading.Thread(target=make_burger, args=(i,))
order_queue.append(task)
task.start()
for task in order_queue:
task.join()
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
在 main 的第一个 for 循环中，任务在不同的线程中创建并启动。 第二个 for 循环确保所有汉堡都制作完成后，程序才能继续运行（即返回 main 之前）。
当我们只有两名员工时，情况更加复杂。 每名员工都由一个线程表示，他/她从一个存储所有任务的普通列表中获取任务。
import threading
import time
order_queue = []
def take_order():
for i in range(3):
order_queue.append(make_burger(i))
def make_burger(order_num):
def making_burger():
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
return making_burger
def working():
while len(order_queue) > 0:
print(f"{threading.current_thread().name} is working...")
task = order_queue.pop(0)
task()
print(f"{threading.current_thread().name} finish task...")
def main():
take_order()
staff1 = threading.Thread(target=working, name="John")
staff1.start()
staff2 = threading.Thread(target=working, name="Jane")
staff2.start()
staff1.join()
staff2.join()
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
运行上面的代码时，其中一个线程可能会出现错误，指示它正在尝试从空列表中获取任务。 您可能想知道为什么会出现这种情况，因为 while 循环中已经有一个条件，只有 task_queue 不为空时才会继续执行。 但还是出错了，这是因为我们遇到了竞争条件。
竞争条件
当多个线程同时尝试访问同一资源或数据时，就会发生竞争条件，导致系统出现问题。 访问资源的时间和顺序对程序逻辑至关重要，不可预测的时间或多个线程交叉访问和修改共享数据可能导致错误。
为了解决程序中的竞争条件，我们将向 task_queue 部署一个锁：
queue_lock = threading.Lock()
为了奏效，我们需要确保在检查队列长度以及从队列中获取任务时拥有访问权限。 当我们拥有权限时，其他线程无法访问队列：
def working():
while True:
with queue_lock:
if len(order_queue) == 0:
return
else:
task = order_queue.pop(0)
print(f"{threading.current_thread().name} is working...")
task()
print(f"{threading.current_thread().name} finish task...")
```
Based on what we have learned so far, we can complete our final code with threading like this:
```
import logging
import threading
import time
logger = logging.getLogger(__name__)
logging.basicConfig(filename="pyburger_threads.log", level=logging.INFO)
queue_lock = threading.Lock()
task_queue = []
order_num = 0
closing = False
def take_order():
global order_num, closing
try:
order_num += 1
logger.info(f"Taking Order #{order_num:04d}...")
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = input("Number of burgers:")
for i in range(int(burger_num)):
with queue_lock:
task_queue.append(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = input("Number of fries:")
for i in range(int(fries_num)):
with queue_lock:
task_queue.append(make_fries(f"{order_num:04d}-fries{i:02d}"))
logger.info(f"Order #{order_num:04d} queued.")
print(f"Order #{order_num:04d} queued, please wait.")
with queue_lock:
task_queue.append(take_order)
except ValueError:
print("Goodbye!")
logger.info("Closing down... stop taking orders and finish all tasks.")
closing = True
def make_burger(order_num):
def making_burger():
logger.info(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
logger.info(f"Burger made #{order_num}")
return making_burger
def make_fries(order_num):
def making_fries():
logger.info(f"Preparing fried #{order_num}...")
time.sleep(2) # time for making fries
logger.info(f"Fries made #{order_num}")
return making_fries
def working():
while True:
with queue_lock:
if len(task_queue) == 0:
if closing:
return
else:
task = None
else:
task = task_queue.pop(0)
if task:
logger.info(f"{threading.current_thread().name} is working...")
task()
logger.info(f"{threading.current_thread().name} finish task...")
else:
time.sleep(1) # rest
def main():
print("Welcome to Pyburger!")
logger.info("Ready for business!")
task_queue.append(take_order)
staff1 = threading.Thread(target=working, name="John")
staff1.start()
staff2 = threading.Thread(target=working, name="Jane")
staff2.start()
staff1.join()
staff2.join()
logger.info("All tasks finished. Closing now.")
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logger.info(f"Orders completed in {elapsed:0.2f} seconds.")
如果您比较使用 asyncio 和 threading 的两个代码段，它们应该会有相似的结果。 您可能想知道哪一个更好，该怎样做出选择。
实际上，编写 asyncio 代码比多线程处理更容易，因为我们不必自己处理潜在的竞争条件和死锁。 控制默认在协程之间传递，因此不需要锁。 不过，Python 线程确实有并行运行的潜力，只是在 GIL 存在的情况下大多数时候无法实现。 我们可以在下一篇博文中讨论 nogil（无线程）Python 时回到这个问题。
从并发中受益
为什么要在编程中使用并发？ 一个主要原因是：速度。 如上文所述，如果我们能缩短等待时间，任务就可以更快完成。 计算中存在不同类型的等待，针对每一种等待，我们倾向于使用不同的方法来节省时间。
I/O 绑定任务
当任务或程序的执行速度主要受限于 I/O 操作（例如从文件或网络读取数据，或等待用户输入）的速度时，该任务或程序就被视为输入/输出 (I/O) 绑定。 I/O 操作通常比其他 CPU 操作慢，因此，涉及大量 I/O 操作的任务可能会花费更多时间。 这些任务的典型示例包括从数据库读取数据、处理 Web 请求或处理大型文件。
使用 async/await 并发可以疏通处理序列，并让其他任务在等待期间得到处理，从而有助于优化 I/O 绑定任务的等待时间。
async/await 并发在许多 Python 应用程序中都非常有用，例如涉及与数据库进行大量通信和处理 Web 请求的 Web 应用程序。 GUI（图形用户界面）也可以从 async/await 并发中受益，允许在用户与应用程序交互时执行后台任务。
CPU 绑定任务
当任务或程序的执行速度主要受限于 CPU 的速度时，该任务或程序就被视为 CPU 绑定。 典型示例包括图像或视频处理，例如调整大小或编辑，以及复杂的数学计算，例如矩阵乘法或训练机器学习模型。
与 I/O 绑定任务不同，CPU 绑定任务很少能够通过使用 async/await 并发优化，因为 CPU 已经在忙于处理这些任务。 如果您的机器有多个 CPU，或者您可以将部分任务卸载到一个或多个 GPU，那么可以创建更多线程并执行多处理来更快完成 CPU 绑定任务。 多处理可以优化这些 CPU 和 GPU 的使用方式，正因如此，如今许多机器学习和 AI 模型都在多个 GPU 上训练。
不过，这很难通过纯 Python 代码实现，因为 Python 本身就是为提供抽象层而设计，让用户无需控制低级别计算过程。 此外，Python 的 GIL 限制了计算机上多个线程之间 Python 资源的共享。 最近，Python 3.13 允许移除 GIL 以实现真正的多线程处理。 我们将在下一篇博文中讨论 GIL 以及如何将其移除。
有时，我们上面提到的方法都无法充分加速 CPU 绑定任务。 在这种情况下，CPU 绑定任务可能需要被分解成更小的任务，以便在多个线程、多个处理器甚至多台机器上同时执行。 这就是并行处理，您可能需要完全重写代码才能实现。 在 Python 中，multiprocessing 软件包提供本地和远程并发，可以用来绕过 GIL 的限制。 我们也将在下一篇博文中展示一些相关示例。
在 PyCharm 中调试并发代码
调试异步或并发代码可能很困难，因为程序不是按顺序执行，这意味着很难看到代码的执行位置和时间。 许多开发者使用 print 帮助追踪代码流，但不推荐这种方式，因为它非常笨拙，用它来研究复杂的程序（例如并发程序）并不容易。 而且，后续整理也很麻烦。
许多 IDE 提供调试器，它们非常适合检查变量和程序流。 调试器还能提供跨多个线程的清晰堆栈跟踪。 来看看如何在 PyCharm 中跟踪示例餐厅模拟的 task_queue。
首先，我们将在代码中设置一些断点。 为此，请点击您希望调试器暂停的行号。 行号将变成红点，表示这里设置了断点。 我们将在第 23、27 和 65 行设置断点，在这些行中，task_queue 在不同的线程中被更改。
然后，我们点击右上角的小虫子图标，以调试模式运行程序。
点击图标后，将打开 Debug（调试）窗口。 程序开始运行，直到代码中高亮显示的第一个断点。
在这里，我们看到 John 线程正在尝试接收任务，第 65 行被高亮显示。 此时，高亮显示的行尚未执行。 当我们想在进入断点之前检查变量时，这很有用。
我们检查一下 task_queue 中的内容。 在 Debug（调试）窗口中开始输入即可，如下所示。
选择或输入“task_queue”，然后按 Enter。 您将看到 take_order 任务在队列中。
现在，我们点击 Step in（步入）按钮执行断点，如下所示。
按下按钮并查看过弹出的 Special Variables（特殊变量）窗口后，我们看到任务变量现在为 John 线程中的 take_order。
再次查询 task_queue 时，我们发现列表现在为空。
点击 Resume Program（恢复程序）按钮，让程序运行。
当程序到达用户输入部分时，PyCharm 会跳转到 Console（控制台）窗口，让我们提供输入。 假设我们想要两个汉堡。 输入“2”，然后按 Enter。
现在，我们到达第二个断点。 如果点击 Threads & Variables（线程和变量）回到该窗口，我们将看到 burger_num 为 2，与我们输入的一样。
现在，我们步入断点，检查 task_queue，就像之前一样。 我们可以看到新增了一个 make_burger 任务。
让程序再次运行，如果我们在程序停止时步入断点，就可以看到 Jane 正在接手任务。
您可以自行检查其余代码。 完成后，按窗口顶部的红色 Stop（停止）按钮即可。
使用 PyCharm 中的调试器，您可以轻松跨线程跟踪程序的执行情况并检查不同的变量。
结论
现在，我们已经学习了 Python 中并发的基础知识，希望您能够通过实践练习加以掌握。 在下一篇博文中，我们将探讨 Python GIL、它的作用以及它缺失时会有什么变化。
PyCharm 为处理并发 Python 代码提供了强大的工具。 如本文所示，调试器允许对异步和线程代码的逐步检查，帮助您跟踪执行流、监控共享资源和检测问题。 PyCharm 拥有直观的断点、实时变量视图、用户输入的无缝控制台集成以及强大的日志记录支持，让编写、测试和调试应用程序变得更加轻松、可靠和清晰。
本博文英文原作者：
Subscribe to PyCharm Blog updates
Discover more
Faster Python: Concurrency in async/await and threading
If you have been coding with Python for a while, especially if you have been using frameworks and libraries such as FastAPI and discord.py, then you have probably been using async/await or asyncio. You may have heard statements like “multithreading in Python isn’t real”, and you may also know about the famous (or infamous) GIL in Python. In light of the denial about multithreading in Python, you might be wondering what the difference between async/await and multithreading actually is – especially in Python programming. If so, this is the blog post for you!
What is multithreading?
In programming, multithreading refers to the ability of a program to execute multiple sequential tasks (called threads) concurrently. These threads can run on a single processor core or across multiple cores. However, due to the limitation of the Global Interpreter Lock (GIL), multithreading in Python is only processed on a single core. The exception is nogil (also called thread-free) Python, which removes the GIL and will be covered in part 2 of this series. For this blog post, we will assume that the GIL is always present.
What is concurrency?
Concurrency in programming means that the computer is doing more than one thing at a time, or seems to be doing more than one thing at a time, even if the different tasks are executed on a single processor. By managing resources and interactions between different parts of a program, different tasks are allowed to make progress independently and in overlapping time intervals.
Both asyncio and threading appear concurrent in Python
Loosely speaking, both the asyncio and threading Python libraries enable the appearance of concurrency. However, your CPUs are not doing multiple things at the exact same time. It just seems like they are.
Imagine you are hosting a multi-course dinner for some guests. Some of the dishes take time to cook, for example, the pie that needs to be baked in the oven or the soup simmering on the stove. While we are waiting for those to cook, we do not just stand around and wait. We will do something else in the meantime. This is similar to concurrency in Python. Sometimes your Python process is waiting for something to get done. For example, some input/output (I/O) processes are being handled by the operating system, and in this time the Python process is just waiting. We can then use async to let another Python process run while it waits.
The difference is who is in charge
If both asyncio and threading appear concurrent, what is the difference between them? Well, the main difference is a matter of who is in charge of which process is running and when. For async/await, the approach is sometimes called cooperative concurrency. A coroutine or future gives up its control to another coroutine or future to let others have a go. On the other hand, in threading, the operating system’s manager will be in control of which process is running.
Cooperative concurrency is like a meeting with a microphone being passed around for people to speak. Whoever has the microphone can talk, and when they are done or have nothing else to say, they will pass the microphone to the next person. In contrast, multithreading is a meeting where there is a chairperson who will determine who has the floor at any given time.
Writing concurrent code in Python
Let’s have a look at how concurrency works in Python by writing some example code. We will create a fast food restaurant simulation using both asyncio and threading.
How async/await works in Python
The asyncio package was introduced in Python 3.4, while the async and await keywords were introduced in Python 3.5. One of the main things that make async/await possible is the use of coroutines. Coroutines in Python are actually generators repurposed to be able to pause and pass back to the main function.
Now, imagine a burger restaurant where only one staff member is working. The orders are prepared according to a first-in-first-out queue, and no async operations can be performed:
import time
def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
def main():
for i in range(3):
make_burger(i)
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
This will take a while to finish:
Preparing burger #0... Burger made #0 Preparing burger #1... Burger made #1 Preparing burger #2... Burger made #2 Orders completed in 15.01 seconds.
Now, imagine the restaurant brings in more staff, so that it can perform work concurrently:
import asyncio
import time
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def main():
order_queue = []
for i in range(3):
order_queue.append(make_burger(i))
await asyncio.gather(*(order_queue))
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
We see the difference between the two:
Preparing burger #0... Preparing burger #1... Preparing burger #2... Burger made #0 Burger made #1 Burger made #2 Orders completed in 5.00 seconds.
Using the functions provided by asyncio, like run and gather, and the keywords async and await, we have created coroutines that can make burgers concurrently.
Now, let’s take a step further and create a more complicated simulation. Imagine we only have two workers, and we can only make two burgers at a time.
import asyncio
import time
order_queue = asyncio.Queue()
def take_order():
for i in range(3):
order_queue.put_nowait(make_burger(i))
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while order_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await order_queue.get()
await task
print(f"{self.name} finished a task...")
async def main():
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
take_order()
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
Here we will use a queue to hold the tasks, and the staff will pick them up.
John is working... Preparing burger #0... Jane is working... Preparing burger #1... Burger made #0 John finished a task... John is working... Preparing burger #2... Burger made #1 Jane finished a task... Burger made #2 John finished a task... Orders completed in 10.00 seconds.
In this example, we use asyncio.Queue to store the tasks, but it will be more useful if we have multiple types of tasks, as shown in the following example.
import asyncio
import time
task_queue = asyncio.Queue()
order_num = 0
async def take_order():
global order_num
order_num += 1
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = input("Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = input("Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
print(f"Order #{order_num:04d} queued.")
await task_queue.put(take_order())
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def make_fries(order_num):
print(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
print(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await task_queue.get()
await task
print(f"{self.name} finish task...")
else:
await asyncio.sleep(1) #rest
async def main():
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
In this example, there are multiple tasks, including making fries, which takes less time, and taking orders, which involves getting input from the user.
Notice that the program stops waiting for the user’s input, and even the other staff who are not taking the order stop working in the background. This is because the input function is not async and therefore is not awaited. Remember, control in async code is only released when it is awaited. To fix that, we can replace:
input("Number of burgers:")
With
await asyncio.to_thread(input, "Number of burgers:")
And we do the same for fries – see the code below. Note that now the program runs in an infinite loop. If we need to stop it, we can deliberately crash the program with an invalid input.
import asyncio
import time
task_queue = asyncio.Queue()
order_num = 0
async def take_order():
global order_num
order_num += 1
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = await asyncio.to_thread(input, "Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = await asyncio.to_thread(input, "Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
print(f"Order #{order_num:04d} queued.")
await task_queue.put(take_order())
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def make_fries(order_num):
print(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
print(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
print(f"{self.name} is working...")
task = await task_queue.get()
await task
print(f"{self.name} finish task...")
else:
await asyncio.sleep(1) #rest
async def main():
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
await asyncio.gather(staff1.working(), staff2.working())
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
By using asyncio.to_thread, we have put the input function into a separate thread (see this reference). Do note, however, that this trick only unblocks I/O-bounded tasks if the Python GIL is present.
If you run the code above, you may also see that the standard I/O in the terminal is scrambled. The user I/O and the record of what is happening should be separate. We can put the record into a log to inspect later.
import asyncio
import logging
import time
logger = logging.getLogger(__name__)
logging.basicConfig(filename='pyburger.log', level=logging.INFO)
task_queue = asyncio.Queue()
order_num = 0
closing = False
async def take_order():
global order_num, closing
try:
order_num += 1
logger.info(f"Taking Order #{order_num:04d}...")
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = await asyncio.to_thread(input, "Number of burgers:")
for i in range(int(burger_num)):
await task_queue.put(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = await asyncio.to_thread(input, "Number of fries:")
for i in range(int(fries_num)):
await task_queue.put(make_fries(f"{order_num:04d}-fries{i:02d}"))
logger.info(f"Order #{order_num:04d} queued.")
print(f"Order #{order_num:04d} queued, please wait.")
await task_queue.put(take_order())
except ValueError:
print("Goodbye!")
logger.info("Closing down... stop taking orders and finish all tasks.")
closing = True
async def make_burger(order_num):
logger.info(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
logger.info(f"Burger made #{order_num}")
async def make_fries(order_num):
logger.info(f"Preparing fries #{order_num}...")
await asyncio.sleep(2) # time for making fries
logger.info(f"Fries made #{order_num}")
class Staff:
def __init__(self, name):
self.name = name
async def working(self):
while True:
if task_queue.qsize() > 0:
logger.info(f"{self.name} is working...")
task = await task_queue.get()
await task
task_queue.task_done()
logger.info(f"{self.name} finish task.")
elif closing:
return
else:
await asyncio.sleep(1) #rest
async def main():
global task_queue
task_queue.put_nowait(take_order())
staff1 = Staff(name="John")
staff2 = Staff(name="Jane")
print("Welcome to Pyburger!")
logger.info("Ready for business!")
await asyncio.gather(staff1.working(), staff2.working())
logger.info("All tasks finished. Closing now.")
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
logger.info(f"Orders completed in {elapsed:0.2f} seconds.")
In this final code block, we have logged the simulation information in pyburger.log and reserved the terminal for messages for customers. We also catch invalid input during the ordering process and switch a closing flag to True if the input is invalid, assuming the user wants to quit. Once the closing flag is set to True, the worker will return, ending the coroutine’s infinite while loop.
How does threading work in Python?
In the example above, we put an I/O-bound task into another thread. You may wonder if we can put all tasks into separate threads and let them run concurrently. Let’s try using threading instead of asyncio.
Consider the code we have as shown below, where we create burgers concurrently with no limitation put in place:
import asyncio
import time
async def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
await asyncio.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
async def main():
order_queue = []
for i in range(3):
order_queue.append(make_burger(i))
await asyncio.gather(*(order_queue))
if __name__ == "__main__":
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
Instead of creating async coroutines to make the burgers, we can just send functions down different threads like this:
import threading
import time
def make_burger(order_num):
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
def main():
order_queue = []
for i in range(3):
task = threading.Thread(target=make_burger, args=(i,))
order_queue.append(task)
task.start()
for task in order_queue:
task.join()
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
In the first for loop in main, tasks are created in different threads and get a kickstart. The second for loop makes sure all the burgers are made before the program moves on (that is, before it returns to main).
It is more complicated when we have only two staff members. Each member of the staff is represented with a thread, and they will take tasks from a normal list where they are all stored.
import threading
import time
order_queue = []
def take_order():
for i in range(3):
order_queue.append(make_burger(i))
def make_burger(order_num):
def making_burger():
print(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
print(f"Burger made #{order_num}")
return making_burger
def working():
while len(order_queue) > 0:
print(f"{threading.current_thread().name} is working...")
task = order_queue.pop(0)
task()
print(f"{threading.current_thread().name} finish task...")
def main():
take_order()
staff1 = threading.Thread(target=working, name="John")
staff1.start()
staff2 = threading.Thread(target=working, name="Jane")
staff2.start()
staff1.join()
staff2.join()
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
print(f"Orders completed in {elapsed:0.2f} seconds.")
When you run the code above, an error may occur in one of the threads, saying that it is trying to get a task from an empty list. You may wonder why this is the case, since we have a condition in the while loop that causes it to continue only if the task_queue is not empty. Nevertheless, we still get an error because we have encountered race conditions.
Race conditions
Race conditions can occur when multiple threads attempt to access the same resource or data at the same time and cause problems in the system. The timing and order of when the resource is accessed are important to the program logic, and unpredictable timing or the interleaving of multiple threads accessing and modifying shared data can cause errors.
To solve the race condition in our program, we will deploy a lock to the task_queue:
queue_lock = threading.Lock()
For working, we need to make sure we have access rights to the queue when checking its length and getting tasks from it. While we have the rights, other threads cannot access the queue:
def working():
while True:
with queue_lock:
if len(order_queue) == 0:
return
else:
task = order_queue.pop(0)
print(f"{threading.current_thread().name} is working...")
task()
print(f"{threading.current_thread().name} finish task...")
```
Based on what we have learned so far, we can complete our final code with threading like this:
```
import logging
import threading
import time
logger = logging.getLogger(__name__)
logging.basicConfig(filename="pyburger_threads.log", level=logging.INFO)
queue_lock = threading.Lock()
task_queue = []
order_num = 0
closing = False
def take_order():
global order_num, closing
try:
order_num += 1
logger.info(f"Taking Order #{order_num:04d}...")
print(f"Order burger and fries for order #{order_num:04d}:")
burger_num = input("Number of burgers:")
for i in range(int(burger_num)):
with queue_lock:
task_queue.append(make_burger(f"{order_num:04d}-burger{i:02d}"))
fries_num = input("Number of fries:")
for i in range(int(fries_num)):
with queue_lock:
task_queue.append(make_fries(f"{order_num:04d}-fries{i:02d}"))
logger.info(f"Order #{order_num:04d} queued.")
print(f"Order #{order_num:04d} queued, please wait.")
with queue_lock:
task_queue.append(take_order)
except ValueError:
print("Goodbye!")
logger.info("Closing down... stop taking orders and finish all tasks.")
closing = True
def make_burger(order_num):
def making_burger():
logger.info(f"Preparing burger #{order_num}...")
time.sleep(5) # time for making the burger
logger.info(f"Burger made #{order_num}")
return making_burger
def make_fries(order_num):
def making_fries():
logger.info(f"Preparing fried #{order_num}...")
time.sleep(2) # time for making fries
logger.info(f"Fries made #{order_num}")
return making_fries
def working():
while True:
with queue_lock:
if len(task_queue) == 0:
if closing:
return
else:
task = None
else:
task = task_queue.pop(0)
if task:
logger.info(f"{threading.current_thread().name} is working...")
task()
logger.info(f"{threading.current_thread().name} finish task...")
else:
time.sleep(1) # rest
def main():
print("Welcome to Pyburger!")
logger.info("Ready for business!")
task_queue.append(take_order)
staff1 = threading.Thread(target=working, name="John")
staff1.start()
staff2 = threading.Thread(target=working, name="Jane")
staff2.start()
staff1.join()
staff2.join()
logger.info("All tasks finished. Closing now.")
if __name__ == "__main__":
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logger.info(f"Orders completed in {elapsed:0.2f} seconds.")
If you compare the two code snippets using asyncio and threading, they should have similar results. You may wonder which one is better and why you should choose one over the other.
Practically, writing asyncio code is easier than multithreading because we don’t have to take care of potential race conditions and deadlocks by ourselves. Controls are passed around coroutines by default, so no locks are needed. However, Python threads do have the potential to run in parallel, just not most of the time with the GIL in place. We can revisit this when we talk about nogil (thread-free) Python in the next blog post.
Benefiting from concurrency
Why do we want to use concurrency in programming? There’s one main reason: speed. Like we have illustrated above, tasks can be completed faster if we can cut down the waiting time. There are different types of waiting in computing, and for each one, we tend to use different methods to save time.
I/O-bound tasks
A task or program is considered input/output (I/O) bound when its execution speed is primarily limited by the speed of I/O operations, such as reading from a file or network, or waiting for user input. I/O operations are generally slower than other CPU operations, and therefore, tasks that involve lots of them can take significantly more time. Typical examples of these tasks include reading data from a database, handling web requests, or working with large files.
Using async/await concurrency can help optimize the waiting time during I/O-bound tasks by unblocking the processing sequence and letting other tasks be taken care of while waiting.
Async/await concurrency is beneficial in many Python applications, such as web applications that involve a lot of communication with databases and handling web requests. GUIs (graphical user interfaces) can also benefit from async/await concurrency by allowing background tasks to be performed while the user is interacting with the application.
CPU-bound tasks
A task or program is considered CPU-bound when its execution speed is primarily limited by the speed of the CPU. Typical examples include image or video processing, like resizing or editing, and complex mathematical calculations, such as matrix multiplication or training machine learning models.
Contrary to I/O-bound tasks, CPU-bound tasks can rarely be optimised by using async/await concurrency, as the CPU is already busy working on the tasks. If you have more than one CPU in your machine, or if you can offload some of these tasks to one or more GPUs, then CPU-bound tasks can be finished faster by creating more threads and performing multiprocessing. Multiprocessing can optimise how these CPUs and GPUs are used, which is also why many machine learning and AI models these days are trained on multiple GPUs.
This, however, is tough to perform with pure Python code, as Python itself is designed to provide abstract layers so users do not have to control the lower-level computation processes. Moreover, Python’s GIL limits the sharing of Python resources across multiple threads on your computer. Recently, Python 3.13 made it possible to remove the GIL, allowing for true multithreading. We will discuss the GIL, and the ability to go without it, in the next blog post.
Sometimes, none of the methods we mentioned above are able to speed up CPU-bound tasks sufficiently. When that is the case, the CPU-bound tasks may need to be broken into smaller ones so that they can be performed simultaneously over multiple threads, multiple processors, or even multiple machines. This is parallel processing, and you may have to rewrite your code completely to implement it. In Python, the multiprocessing package offers both local and remote concurrency, which can be used to work around the limitation of the GIL. We will also look at some examples of that in the next blog post.
Debugging concurrent code in PyCharm
Debugging async or concurrent code can be hard, as the program is not executed in sequence, meaning it is hard to see where and when the code is being executed. Many developers use print to help trace the flow of the code, but this approach is not recommended, as it is very clumsy and using it to investigate a complex program, like a concurrent one, isn’t easy. Plus, it is messy to tidy up after.
Many IDEs provide debuggers, which are great for inspecting variables and the flow of the program. Debuggers also provide a clear stack trace across multiple threads. Let’s see how we can track the task_queue of our example restaurant simulation in PyCharm.
First, we will put down some breakpoints in our code. You can do that by clicking the line number of the line where you want the debugger to pause. The line number will turn into a red dot, indicating that a breakpoint is set there. We will put breakpoints at lines 23, 27, and 65, where the task_queue is changed in different threads.
Then we can run the program in debug mode by clicking the little bug icon in the top right.
After clicking on the icon, the Debug window will open up. The program will run until it hits the first breakpoint highlighted in the code.
Here we see the John thread is trying to pick up the task, and line 65 is highlighted. At this point, the highlighted line has not been executed yet. This is useful when we want to inspect the variables before entering the breakpoint.
Let’s inspect what’s in the task_queue. You can do so simply by starting to type in the Debug window, as shown below.
Select or type in “task_queue”, and then press Enter. You will see that the take_order task is in the queue.
Now, let’s execute the breakpoint by clicking the Step in button, as shown below.
After pressing that and inspecting the Special Variables window that pops up, we see that the task variable is now take_order in the John thread.
When querying the task_queue again, we see that now the list is empty.
Now let’s click the Resume Program button and let the program run.
When the program hits the user input part, PyCharm will bring us to the Console window so we can provide the input. Let’s say we want two burgers. Type “2” and press Enter.
Now we hit the second breakpoint. If we click on Threads & Variables to go back to that window, we’ll see that burger_num is two, as we entered.
Now let’s step into the breakpoint and inspect the task_queue, just like we did before. We see that one make_burger task has been added.
We let the program run again, and if we step into the breakpoint when it stops, we see that Jane is picking up the task.
You can inspect the rest of the code yourself. When you are done, simply press the red Stop button at the top of the window.
With the debugger in PyCharm, you can follow the execution of your program across different threads and inspect different variables very easily.
Conclusion
Now we have learned the basics of concurrency in Python, and I hope you will be able to master it with practice. In the next blog post, we will have a look at the Python GIL, the role it plays, and what changes when it is absent.
PyCharm provides powerful tools for working with concurrent Python code. As demonstrated in this blog post, the debugger allows the step-by-step inspection of both async and threaded code, helping you track the execution flow, monitor shared resources, and detect issues. With intuitive breakpoints, real-time variable views, seamless console integration for user input, and robust logging support, PyCharm makes it easier to write, test, and debug applications with confidence and clarity.