Python

关于Python协程

Posted by FanHao on 2023-08-18

问题引入

CPU的速度远远快于磁盘、网络等IO。在一个线程中，CPU执行代码的速度极快，然而，一旦遇到IO操作，如读写文件、发送网络数据时，就需要等待IO操作完成，才能继续进行下一步操作。这种情况称为同步IO。
在IO操作的过程中，当前线程被挂起，而其他需要CPU执行的代码就无法被当前线程执行了。
因为一个IO操作就阻塞了当前线程，导致其他代码无法执行，所以我们必须使用多线程或者多进程来并发执行代码，为多个用户服务。每个用户都会分配一个线程，如果遇到IO导致线程被挂起，其他用户的线程不受影响。
多线程和多进程的模型虽然解决了并发问题，但是系统不能无上限地增加线程。由于系统切换线程的开销也很大，所以，一旦线程数量过多，CPU的时间就花在线程切换上了，真正运行代码的时间就少了，结果导致性能严重下降。
由于我们要解决的问题是CPU高速执行能力和IO设备的龟速严重不匹配，多线程和多进程只是解决这一问题的一种方法。
另一种解决IO问题的方法是异步IO。当代码需要执行一个耗时的IO操作时，它只发出IO指令，并不等待IO结果，然后就去执行其他代码了。一段时间后，当IO返回结果时，再通知CPU进行处理。

异步IO

异步IO模型需要一个消息循环，在消息循环中，主线程不断地重复“读取消息-处理消息”这一过程：如下伪代码

1
2
3
4
loop = get_event_loop()
while True:
    event = loop.get_event()
    process_event(event)

消息模型是如何解决同步IO必须等待IO操作这一问题的呢？当遇到IO操作时，代码只负责发出IO请求，不等待IO结果，然后直接结束本轮消息处理，进入下一轮消息处理过程。当IO操作完成后，将收到一条“IO完成”的消息，处理该消息时就可以直接获取IO操作结果。
在“发出IO请求”到收到“IO完成”的这段时间里，同步IO模型下，主线程只能挂起，但异步IO模型下，主线程并没有休息，而是在消息循环中继续处理其他消息。这样，在异步IO模型下，一个线程就可以同时处理多个IO请求，并且没有切换线程的操作。对于大多数IO密集型的应用程序，使用异步IO将大大提升系统的多任务处理能力。

举例：桌面应用程序处理所有键盘和鼠标操作都类似这种设计模型。某些时候GUI线程处理某一个鼠标操作非常慢时，用户即会感觉程序界面卡死停止响应。

协程

协程是子例程的更一般形式，适用于高并发、IO密集型任务，不适用于计算密集型任务。子例程可以在某一点进入并在另一点退出。协程则可以在许多不同的点上进入、退出和恢复。它们可通过async def 语句来实现。
协程不是计算机内部提供的，不像进程、线程，由电脑本身提供，它是由程序员人为创造的, 实现函数异步执行。
协程(Coroutine)，也可以被称为微线程，是一种用户太内的上下文切换技术，其实就是通过一个线程实现代码块相互切换执行。看上去像子程序，但执行过程中，在子程序内部可中断，然后转而执行别的子程序，在适当的时候再返回来接着执行

代码示例

以下代码通过async关键实现了2个异步函数分别处理两个任务：
若按照正常流程编写代码执行。将花费2.5s时间。使用异步协程方式执行，则总计花费时间1.5s，分析如下：
①执行say_hi，遇到IO任务阻塞1s，此时cpu不会停留等待IO完成而是跳转执行say_hello函数。
②执行完成say_hello函数后，CPU进入等待，say_hi函数IO执行完成，cpu立即执行余下工作。然后再等待say_hello()函数IO完成。
③因为CPU在执行IO耗时任务时没有等待处于空闲状态，而是依然在执行其他任务。此时CPU利用率非常高。执行完成两个任务的时间为1.5秒

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import time
import asyncio


async def say_hello():
    # IO任务1；休眠1s模拟IO操作
    a = time.time()
    print("Hello")
    await asyncio.sleep(1.5)
    b = time.time()
    print("world!", b - a)

async def say_hi():
    # IO任务2；休眠1.5模拟IO操作
    a = time.time()
    print("hi")
    await asyncio.sleep(1)
    b = time.time()
    print("world", b - a)


if __name__ == "__main__":
    start = time.time()
    tasks = [
        asyncio.ensure_future(fn) for fn in [say_hi(), say_hello()]
    ]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    print(time.time()-start)

以上创建事件循环对象、将携程对象封装task任务、运行task对象代码也改写如下：
1
2
3
4
5
6
loop = asyncio.get_event_loop()
task = [
    loop.create_task(fn) for fn in [say_hi(), say_hello()]
]
loop.run_until_complete(asyncio.wait(task))
loop.close()

示例二：获取Web网站的主页，并打印头部
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
async def wget(host):
    # 获取主页
    print("wget %s..." % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = await connect
    header = 'GET / HTTP/1.O\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    await writer.drain()
    while True:
        line = await reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    writer.close()


if __name__ == "__main__":
    start = time.time()
    tasks = [
        wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']
    ]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()
    print("总花费时间：%s" % (time.time()-start))

FEATURED TAGS
Python
FRIENDS