Python asyncio Module

asynciois a module in the Python standard library, used for writing asynchronous I/O operation code.

asyncio provides an efficient way to handle concurrent tasks, especially suitable for I/O-intensive operations such as network requests, file reading and writing, etc.

By usingasyncio, you can process multiple tasks simultaneously in a single thread without using multiple threads or processes.

Why do we need asyncio?

In traditional synchronous programming, when a task needs to wait for an I/O operation (such as a network request) to complete, the program blocks until the operation completes. This leads to low efficiency, especially when dealing with a large number of I/O operations.

asyncioBy introducing the asynchronous programming model, the program can continue executing other tasks while waiting for I/O operations, thereby improving the concurrency and efficiency of the program.

Imagine you are running a restaurant:

  • Synchronous mode (ordinary functions):You only have one chef. Guest A orders a steak, and the chef starts to fry the steak (this requires waiting 5 minutes). During these 5 minutes of frying the steak, the chef is completely occupied and cannot do anything else. Even if guest B only wants a glass of water, he has to wait.
  • Asynchronous mode (asyncio):You have multiple chefs (actually still one, but very smart). After the chef starts frying Guest A's steak, he realizes he needs to wait. He immediately marks the steak as waiting, then turns around to pour water for Guest B. After pouring the water, he comes back to check if the steak is almost done. If it's not done yet, he can go handle Guest C's order. In this way, during the time of waiting for I/O (such as frying a steak, network requests, reading and writing files), the chef (CPU) is always working efficiently.

asyncio is the standard library that Python uses to implement this smart working mode. It allows you to write single-threaded concurrent code, especially suitable for I/O-intensive scenarios such as web crawlers, web servers, microservices, etc.

Its core is the event loop, coroutines, and tasks.

Core Concepts of asyncio

1. Coroutine

A coroutine isasyncioone of the core concepts. It is a special function that can pause during execution and resume later. Coroutines are defined using theasync defkeyword and, through theawaitkeyword, pause execution to wait for an asynchronous operation to complete.

Example

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

2. Event Loop

The event loop isasynciothe core component, responsible for scheduling and executing coroutines. It continuously checks whether there are tasks to execute and calls corresponding callback functions after tasks complete.

Example

async def main():
    await say_hello()

asyncio.run(main())

3. Task

A task is a wrapper around a coroutine, representing a coroutine that is executing or will be executed. You can use theasyncio.create_task()function to create a task and add it to the event loop.

Example

async def main():
    task = asyncio.create_task(say_hello())
    await task

4. Future

Futureis an object representing the result of an asynchronous operation. It is usually used in low-level APIs to represent an operation that has not yet completed. You can use theawaitkeyword to wait forFutureto complete.

Example

async def main():
    future = asyncio.Future()
    await future

Basic Usage and Code Examples

Let's understand the above concepts through a classic example of concurrently accessing multiple URLs.

Suppose we need to fetch the content of three different URLs. Using a synchronous approach, tasks execute sequentially, and the total time is the sum of the three request times. Withasyncio, we can send these three requests at the same time, and the total time is close to the slowest request.

Synchronous Version (for comparison)

Example

import time
import requests

def fetch_url(url):
    """Simulate a time-consuming network request (synchronous version)"""
    print(f"Start fetching: {url}")
    time.sleep(2)  # Simulate 2 seconds of network delay
    print(f"Finished fetching: {url}")
    return f"Data from {url}"

def main_sync():
    urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3']
    results = []
    start = time.time()
   
    for url in urls:
        result = fetch_url(url)  # Must wait for the previous one to complete before starting the next
        results.append(result)
   
    end = time.time()
    print(f"Synchronous version total time: {end - start:.2f} seconds")
    print(f"Results: {results}")

if __name__ == "__main__":
    main_sync()

Expected output：

开始获取: https://example.com/1
完成获取: https://example.com/1
开始获取: https://example.com/2
完成获取: https://example.com/2
开始获取: https://example.com/3
完成获取: https://example.com/3
同步版本总耗时: 6.00 秒
结果: [‘来自 https://example.com/1 的数据‘, ‘来自 https://example.com/2 的数据‘, ‘来自 https://example.com/2 的数据‘]

It takes about 6 seconds in total.

Asynchronous Version (using asyncio)

We need to useaiohttplibrary to replacerequestsfor making asynchronous HTTP requests. First, install it:pip install aiohttp。

Example

import asyncio
import aiohttp
import time

async def fetch_url_async(session, url):
    """Simulate a time-consuming network request (asynchronous version)"""
    print(f"Start async fetching: {url}")
    # Note: Here we use aiohttp's async get method and wait with await
    async with session.get(url) as response:
        # Simulate that processing the response also takes time
        await asyncio.sleep(2)  # Use asyncio.sleep to simulate I/O waiting; it does not block the thread
        text = await response.text()
        print(f"Finished async fetching: {url}")
        return f"Data from {url} (length: {len(text)})"

async def main_async():
    urls = ['https://httpbin.org/get', 'https://httpbin.org/delay/1', 'https://httpbin.org/headers']
   
    async with aiohttp.ClientSession() as session:  # Create an asynchronous HTTP session
        # Create a task (Task) for each URL
        tasks = []
        for url in urls:
            # create_task adds the coroutine to the event loop and starts scheduling immediately
            task = asyncio.create_task(fetch_url_async(session, url))
            tasks.append(task)
       
        print("All tasks created, starting concurrent execution...")
       
        # Use asyncio.gather to run all tasks concurrently and wait for all of them to complete
        # gather returns a list of results, with the order matching the order of the tasks passed in
        results = await asyncio.gather(*tasks)
       
        return results

if __name__ == "__main__":
    start = time.time()
    # asyncio.run() is a convenient way to start the event loop and run the top-level coroutine
    final_results = asyncio.run(main_async())
    end = time.time()
   
    print(f"\n"Asynchronous version total time: {end - start:.2f} seconds")
    for res in final_results:
        print(res)

Expected output：

所有任务已创建，开始并发执行...
开始异步获取: https://httpbin.org/get
开始异步获取: https://httpbin.org/delay/1
开始异步获取: https://httpbin.org/headers
（大约 2 秒后，所有请求几乎同时完成）
完成异步获取: https://httpbin.org/headers
完成异步获取: https://httpbin.org/get
完成异步获取: https://httpbin.org/delay/1

异步版本总耗时: 2.10 秒  # 注意！总耗时远小于 6 秒
来自 https://httpbin.org/get 的数据 (长度: 274)
来自 https://httpbin.org/delay/1 的数据 (长度: 392)
来自 https://httpbin.org/headers 的数据 (长度: 177)

Code Analysis：

  • async def: defines the coroutine functionfetch_url_asyncandmain_async。
  • await: infetch_url_async, weawait session.get()andawait response.text(), this tells the event loop: "This network request takes time; go execute other ready tasks first."
  • asyncio.create_task(): wraps thefetch_url_asynccoroutine into aTask, enabling it to be scheduled by the event loop and achieve concurrency.
  • asyncio.gather(*tasks): a very practical function that runs all passed coroutines/tasks concurrently, waits for all of them to complete, and finally collects all results.
  • asyncio.run(main_async()): the recommended approach in Python 3.7+, responsible for creating the event loop, running the coroutine, and closing the loop.

Key Functions and Parameter Descriptions

The following table listsasyncioseveral of the most commonly used high-level functions:

Function Main Purpose Common Parameter Descriptions
asyncio.run(coro, *, debug=False) Runs a top-level coroutine and manages the event loop's lifecycle. It is the main entry point of the program. coro: the coroutine object to run.
debug: set toTrueto enable the debug mode of the event loop.
asyncio.create_task(coro, *, name=None) Wraps a coroutine into aTaskobject and queues it in the event loop for scheduling. This is the main way to achieve concurrency. coro: the coroutine object to wrap.
name: (Python 3.8+) specify a name for the task for easier debugging.
asyncio.gather(*aws, return_exceptions=False) Concurrently runsmultiple asynchronous tasks (awscan accept coroutines, tasks, etc.), waits for all to complete, and returns a list of results. *aws: variable arguments, pass in multiple asynchronous objects.
return_exceptions: defaults toFalse, any exception raised by a task will be immediately propagated togatherthe caller. Set toTruewhen set, exceptions will be returned as part of the results.
asyncio.sleep(delay, result=None) Asynchronouslysleeps for the specified number of seconds. This is the key difference fromtime.sleep(blocking). delay: Number of seconds to sleep.
result: The value returned after the sleep ends.
asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED) Run tasks concurrently and wait until the specified condition is met. Returns two sets(done, pending), which are the completed and uncompleted tasks respectively. aws: Collection of asynchronous objects.
timeout: Timeout duration (seconds).
return_when: Determines when to return; options:FIRST_COMPLETED(first completed),FIRST_EXCEPTION(first exception),ALL_COMPLETED(all completed, default).
asyncio.to_thread(func, /, *args, **kwargs) (Python 3.9+) Runs an ordinary, potentially blocking synchronous function in a separate thread and returns anawaitawaitable coroutine. Used for handling CPU-intensive or blocking I/O. func: The synchronous function to run in the thread.
*args, **kwargs: Arguments passed to the function.

Visual Understanding: Asynchronous Task Scheduling Flow

Diagram explanation: This flowchart shows how the event loop works like a dispatcher. It maintains a task queue. When a task executes toawait(e.g., waiting for a network response), it is suspended. The event loop immediately finds the next runnable (ready) task from the queue and executes it. When the I/O operation of the suspended task completes, the event loop receives a notification, changes the task's status back to ready, and continues executing it at some future moment. In this way, during I/O waits, the CPU is fully utilized to execute other tasks, achieving concurrency within a single thread.

Basic Usage of asyncio

1. Running a Coroutine

To run a coroutine, you can use theasyncio.run()function. It creates an event loop and runs the specified coroutine.

Example

import asyncio

async def main():
    print("Start")
    await asyncio.sleep(1)
    print("End")

asyncio.run(main())

2. Executing Multiple Tasks Concurrently

You can use theasyncio.gather()function to run multiple coroutines concurrently and wait for all of them to complete.

Example

import asyncio

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

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

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

3. Timeout Control

You can use theasyncio.wait_for()function to set a timeout for a coroutine. If the coroutine does not complete within the specified time, aasyncio.TimeoutErrorexception will be raised.

Example

import asyncio

async def long_task():
    await asyncio.sleep(10)
    print("Task finished")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=5)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

Application Scenarios of asyncio

asyncioEspecially suitable for the following scenarios:

  1. Network requests: such as HTTP requests, WebSocket communication, etc.
  2. File I/O: such as asynchronous file reading and writing.
  3. Database operations: such as asynchronous database access.
  4. Real-time data processing: such as real-time message queue processing.

Common Classes, Methods, and Functions

1. Core functions

Method/Function Description Example
asyncio.run(coro) Run the asynchronous main function (Python 3.7+) asyncio.run(main())
asyncio.create_task(coro) Create a task and add it to the event loop task = asyncio.create_task(fetch_data())
asyncio.gather(*coros) Run multiple coroutines concurrently await asyncio.gather(task1, task2)
asyncio.sleep(delay) Asynchronous wait (non-blocking) await asyncio.sleep(1)
asyncio.wait(coros) Control how tasks complete done, pending = await asyncio.wait([task1, task2])

2. Event Loop

Method Description Example
loop.run_until_complete(future) Run until the task completes loop.run_until_complete(main())
loop.run_forever() Run the event loop forever loop.run_forever()
loop.stop() Stop the event loop loop.stop()
loop.close() Close the event loop loop.close()
loop.call_soon(callback) Schedule a callback function to run immediately loop.call_soon(print, "Hello")
loop.call_later(delay, callback) Execute a callback with a delay loop.call_later(5, callback)

3. Coroutines and Tasks

Method/Decorator Description Example
@asyncio.coroutine Coroutine decorator (legacy, Python 3.4-3.7) @asyncio.coroutine
def old_coro():
async def Define a coroutine (Python 3.5+) async def fetch():
task.cancel() Cancel a task task.cancel()
task.done() Check whether a task is completed if task.done():
task.result() Get the task result (requires the task to be completed) data = task.result()

4. Synchronization primitives (similar tothreading）

Class Description Example
asyncio.Lock() Asynchronous mutex lock lock = asyncio.Lock()
async with lock:
asyncio.Event() Event notification event = asyncio.Event()
await event.wait()
asyncio.Queue() Asynchronous queue queue = asyncio.Queue()
await queue.put(item)
asyncio.Semaphore() Semaphore sem = asyncio.Semaphore(5)
async with sem:

5. Networking and subprocesses

Method/Class Description Example
asyncio.open_connection() Establish a TCP connection reader, writer = await asyncio.open_connection('host', 80)
asyncio.start_server() Create a TCP server server = await asyncio.start_server(handle, '0.0.0.0', 8888)
asyncio.create_subprocess_exec() Create a subprocess proc = await asyncio.create_subprocess_exec('ls')

6. Utility tools

Method Description Example
asyncio.current_task() Get the current task task = asyncio.current_task()
asyncio.all_tasks() Get all tasks tasks = asyncio.all_tasks()
asyncio.shield(coro) Protect a task from being cancelled await asyncio.shield(critical_task)
asyncio.wait_for(coro, timeout) Wait with a timeout try: await asyncio.wait_for(task, 5)

Example

1. Basic coroutine example

Example

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(hello())  # Python 3.7+

2. Executing tasks concurrently

Example

async def fetch(url):
    print(f"Fetching {url}")
    await asyncio.sleep(2)
    return f"Data from {url}"

async def main():
    results = await asyncio.gather(
        fetch("url1.com"),
        fetch("url2.com")
    )
    print(results)

asyncio.run(main())

3. Using an asynchronous queue

Example

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

Notes

  1. Python version: Some features require Python 3.7+ (such asasyncio.run()）。

  2. Blocking operations: Avoid using synchronous blocking code in coroutines (such astime.sleep()）。

  3. Debugging: Set thePYTHONASYNCIODEBUG=1environment variable to enable debug mode.

  4. Cancelling tasks: A cancelled task will raiseCancelledError, which needs to be handled properly.

Other extensions