Python异步协程总执行时间的数学表达式推导：含阻塞/非阻塞操作

问题描述 #

我正在开发一个Python异步程序，调度三个协程，每个协程包含以下三个操作步骤：

• 耗时X_i秒的阻塞操作（第i个协程记为X_i）；
• 理想情况下不会阻塞其他任务、耗时Y_i秒的非阻塞操作（第i个协程记为Y_i）；
• 耗时Z_i秒的最终阻塞操作（第i个协程记为Z_i）。

请问结合这些单个任务的耗时参数，可用什么数学表达式来表示该程序的总执行时间T？

示例代码 #

import asyncio
import time
import random

async def fetch_data(data: str, delay: int) -> str:
    print('正在获取数据...')
    await asyncio.sleep(delay)
    print(f'数据获取成功！非阻塞延迟 {delay} 秒')
    return data

async def process_data(i: int, block_delay_1st: int, block_delay_2nd: int, non_block_delay: int=2):
    '''
    协程：代表一个包含三步的任务
    1. X秒的阻塞操作
    2. Y秒的可等待非阻塞操作
    3. Z秒的最终阻塞操作
    '''
    print(f"开始处理任务-{i}... 第一步阻塞延迟: {block_delay_1st}, 非阻塞延迟: {non_block_delay}, 第二步阻塞延迟: {block_delay_2nd}")
    start_time = time.time()  # 启动任务计时器
    # 第一步：同步阻塞操作
    time.sleep(block_delay_1st)
    # 第二步：异步非阻塞操作
    await fetch_data("data_str", non_block_delay)
    # 第三步：同步阻塞操作
    time.sleep(block_delay_2nd)
    end_time = time.time()  # 结束任务计时器
    print(f"任务-{i}处理完成！耗时: {end_time - start_time} 秒")

async def main():
    # 随机生成三个任务的阻塞/非阻塞延迟参数
    block_delay_dict = {
        1: (random.choice([1, 2, 3]), random.choice([1, 2, 3]), random.choice([1, 2, 3])),
        2: (random.choice([1, 2, 3]), random.choice([1, 2, 3]), random.choice([1, 2, 3])),
        3: (random.choice([1, 2, 3]), random.choice([1, 2, 3]), random.choice([1, 2, 3]))
    }
    # 创建任务列表
    tasks = [
        process_data(i, delay_1st, delay_2nd, non_block_delay)
        for i, (delay_1st, delay_2nd, non_block_delay) in block_delay_dict.items()
    ]
    # 并发执行所有任务
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    '''
    总执行时间 T ≈ X1 + X2 + X3 + max(Y1, Y2, Y3) + Z1 + Z2 + Z3 ???
    '''
    start = time.perf_counter()
    asyncio.run(main())
    end = time.perf_counter()
    print(f"程序总执行时间: {end - start} 秒")

推导与结论 #

核心前提说明 #

代码中的time.sleep()是同步阻塞操作——它会占据整个asyncio事件循环，直到自身执行完成，期间其他协程完全无法运行。而await asyncio.sleep()是真正的异步非阻塞操作，会主动让出CPU，让其他就绪的协程执行。

总执行时间拆解 #

基于代码的实现逻辑，三个协程的执行流程时间线如下：

1. 第一步阻塞操作（X₁、X₂、X₃）：
   由于time.sleep()的同步阻塞特性，三个协程的这一步会串行执行——事件循环必须等前一个协程的阻塞操作完成后，才能切换到下一个协程。这部分总耗时为所有X_i的和：X₁ + X₂ + X₃。

2. 第二步非阻塞操作（Y₁、Y₂、Y₃）：
   当所有协程都完成第一步后，它们的await asyncio.sleep(Y_i)会并发执行——多个异步延迟可以同时进行，这部分的总耗时由最长的那个Y_i决定：max(Y₁, Y₂, Y₃)。

3. 第三步阻塞操作（Z₁、Z₂、Z₃）：
   和第一步逻辑一致，time.sleep(Z_i)同步阻塞，三个协程的这一步串行执行，总耗时为所有Z_i的和：Z₁ + Z₂ + Z₃。

最终数学表达式 #

总执行时间T的表达式为：
$$ T = (X_1 + X_2 + X_3) + \max(Y_1, Y_2, Y_3) + (Z_1 + Z_2 + Z_3) $$

额外说明：全异步场景 #

如果把代码中的time.sleep()替换为await asyncio.sleep()（让所有操作都异步非阻塞），此时三个协程可以完全并发执行，总执行时间将由单个协程的最长总耗时决定：
$$ T = \max(X_1+Y_1+Z_1,\ X_2+Y_2+Z_2,\ X_3+Y_3+Z_3) $$

内容的提问来源于stack exchange，提问作者xjqian