Python异步编程：核心要义与事件循环及回调机制解析

Python异步编程：核心要义与事件循环及回调机制解析

摘要：

Python异步编程以async/await为基础构建协程，协程在事件循环中运行。当协程生成任务并遇到await时会挂起，待I/O操作完成触发回调后恢复执行，借助事件循环实现非阻塞调度并发任务，从而达成单线程下的高并发效果。

正文

1. 异步编程的基本概念

1.1 并发与并行的差异

首先，来了解顺序、并发、串行和并行这几个基础概念。
– 顺序与并发：任务开启的时机
– ① 顺序：前一个任务完成后，当前任务才得以开启；
– ② 并发：不管前一个任务是否完成，当前任务均可开启。
– 串行与并行：任务执行的状态
– ① 串行：仅有一个任务执行单元，只能单个任务依次执行；
– ② 并行：存在多个任务执行单元，能够多个任务同时执行。
– 并发与并行的区别在于：
– ① 并发着重于具备处理多个任务的能力，关注任务的抽象调度，不一定同时执行多个任务；
– ② 并行着重于拥有同时处理多个任务的能力，关注任务的实际执行情况。

1.2 并发与并行的关联

• （1）并发和并行都可由多个线程构成，若线程能被多个CPU同时执行则为并行，而并发是多个线程被一个CPU轮流切换执行；
• （2）并行需多核CPU支持，并发单核即可实现（并发宏观似并行，微观是串行）；
• （3）并行是并发的子集，并发未必并行，并行必定属于并发。

1.3 异步模型的概念

除顺序和并行执行模型外，还有异步模型。在异步模型中，允许同一时间处理多个事件。程序调用耗时较长的功能时，不会阻塞执行流程，会继续往下运行，功能执行完毕后能获取结果。简言之，异步编程是程序在等待操作（如I/O）完成时仍能执行其他任务的编程方式。

通过示例对比同步与异步：程序通过网络获取两个文件并合并，异步系统中会开启额外线程处理，两线程分别获取文件，无需等待彼此，获取结果后再同步合并。单线程读取文件并做数学运算时，异步系统发起读取请求后，等待读取时可返回CPU做数学运算。异步编程中，耗时功能常配回调函数，功能执行完通过回调返回结果。

1.4 异步编程模型与进程、线程的关系

以多线程为例，多线程可同时并发或并行执行多个指令。单核处理器上，多线程看似并行，实则是处理器通过调度算法切换线程；多核处理器上，线程真正并行，多个处理器同时执行多个线程以高效处理。例如开启两个浏览器窗口同时下载文件，各用新线程，无需等待，并行下载。多线程是异步编程的一种实现形式，异步编程中线程可能随时被挂起替换，程序员可将任务写成可间隔执行的小步骤，利用线程池或进程池解决异步任务。

2. 异步编程的重要实现方式 – 事件循环和回调机制

2.1 事件循环机制

事件循环是异步编程核心，能让程序在处理I/O等耗时操作时保持响应，类比人类“自动驾驶模式”，可在任务间切换不丢重点。计算系统中，事件源产生事件，事件处理者处理事件，事件循环管理事件，循环执行追踪事件顺序放队列，主线程空闲时调用处理者，事件处理完继续下一个。如GUI应用中循环检测用户事件并后台处理数据。

事件循环工作步骤：
– （1）注册事件与回调：程序向事件循环注册感兴趣事件及对应回调函数；
– （2）等待事件发生：事件循环运行等待事件，期间程序执行其他任务；
– （3）事件触发与回调执行：事件发生时，事件循环调用相应回调函数处理。

2.2 回调机制

回调机制是异步编程常见设计模式，任务完成时自动执行预定函数，类似条件反射。技术层面，回调是函数指针或引用，事件发生时由事件处理程序调用，可延迟执行代码部分，解耦事件发生与处理逻辑，使代码模块化。

2.3 具体实现

Python的Asyncio模块提供管理事件、协程、任务和线程的方法及并发编程原语。主要组件有事件循环、协程、Futures、Tasks。Asyncio管理事件循环的方法如下：

定义三个异步任务依次执行，代码示例：

import asyncio
import random
import time

def function_1(end_time, loop):
    print("function_1 called")
    time.sleep(random.randint(0, 1))
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_2, end_time, loop)
    else:
        loop.stop()

def function_2(end_time, loop):
    print("function_2 called ")
    time.sleep(random.randint(0, 1))
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_3, end_time, loop)
    else:
        loop.stop()

def function_3(end_time, loop):
    print("function_3 called")
    time.sleep(random.randint(0, 1))
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, function_1, end_time, loop)
    else:
        loop.stop()

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
end_loop = loop.time() + 9
loop.call_soon(function_1, end_loop, loop)
loop.run_forever()
loop.close()

运行结果如下：

# 此处为运行时的输出结果，根据代码实际运行情况展示