Python 协程原理全面分析

在介绍Python协程原理之前，需要先了解一些概念：

并发：同时处理多个任务。
并行：同时处理多个任务并使它们同时运行。关注于任务的执行，强调在物理上同时运行多个任务。
同步：任务按照一定的顺序进行，只有先完成前面任务才能完成后面任务。
异步：不按照任务排定的先后顺序进行，而是根据情况随时安排执行任务。异步任务可以在等待IO的过程中进行其他任务，从而提高程序效率。

协程是为了处理异步的一种技术，协程比线程的切换更加轻量级，能够处理大量IO操作，提高程序的并发性和性能。

什么是协程

协程（Coroutine）是一种用户态的轻量级线程。用户可以自行控制协程的启动、暂停、恢复和终止。

协程的本质是控制流的上下文切换。

协程的启动、恢复和终止由用户直接控制，不需要操作系统的参与，避免了切换进程或线程的开销，因此协程的切换非常快速，有利于实现高效率的并发程序。

Python协程的实现

Python实现协程的方式分为三种：

生成器（generator）；
asyncio库；
gevent库。

这里我们主要介绍使用生成器实现协程。

生成器实现协程

使用生成器实现协程，需要用到两个关键字：yield 和 send()。

yield 是将一个函数变为生成器函数，可以将代码执行到 yield 处时中断函数，并返回一个值暂停函数；

send() 是从生成器的外部向其中传入一个值，继续执行被中断的代码。

在协程中，yield 和 send() 的搭配使用实现协程的切换。

首先，定义一个协程函数：

def coroutine_func():
    print('Coroutine started.')
    while True:
        x = yield
        print('Got:', x)

在该函数中使用 yield 关键字，使该函数变为生成器函数。

函数中的 yield 实现中断函数并返回一个值，而 x = yield 则将生成器函数定义为一个可接收外部传值的函数。

接下来，我们通过以下步骤使用上面定义的协程函数：

使用 next() 函数启动协程；
使用 send() 函数向协程发送值；
使用 throw() 函数关闭协程。

# 启动生成器
coroutine = coroutine_func()
next(coroutine)

使用 send() 函数向协程发送值：

coroutine.send('Hello')
coroutine.send('World')

最后，关闭协程：

coroutine.throw(StopIteration)

执行结果：

Coroutine started.
Got: Hello
Got: World
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

协程函数被成功启动并执行到第一个 yield 处，接着使用 send() 函数发送了两个值并分别得到输出，最终使用 throw() 函数关闭了协程。

使用协程的优点

避免频繁的IO调用

对于IO密集型的操作（如文件、网络IO等），使用传统的同步IO会导致程序长时间等待，浪费CPU时间的开销。而使用异步IO可以在等待IO操作完成的过程中，继续进行其他操作，从而提高程序的并发性和性能。

修复回调地狱

使用回调函数处理异步操作会导致函数嵌套层数增加，代码越来越难以维护的问题。而使用协程可以避免这种情况。

示例说明

示例一：爬取豆瓣图书数据

以豆瓣图书数据为例，使用协程实现异步爬取数据，加快爬取速度。

首先，我们需要安装 requests、beautifulsoup4 和 fake_useragent 库。

定义一个获取豆瓣图书信息的函数，该函数包含三个协程函数：

fetch_content，获取HTTP响应内容；
parse_html，解析HTML文本；
save_csv，保存数据到CSV文件中。

import csv
import random
import time

import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent


def fetch_content(url):
    headers = {
        'User-Agent': UserAgent().random,
        'Referer': 'https://book.douban.com/',
        'Connection': 'keep-alive'
    }
    response = requests.get(url, headers=headers)
    return response.text


def parse_html(html):
    soup = BeautifulSoup(html, 'lxml')
    for item in soup.select('li.subject-item'):
        uid = item.select_one('div.info label.uid').text.strip()
        title = item.select_one('div.info h2 a').text.strip()
        authors = item.select_one('div.info div.pub').text.strip().split('/')[0]
        rating = item.select_one('div.info div.star span.rating_nums').text.strip()
        yield uid, title, authors, rating


def save_csv(filename, data):
    with open(filename, 'a', encoding='utf-8-sig', newline='') as f:
        writer = csv.writer(f)
        writer.writerows(data)

定义一个协程函数，接收书籍的url和CSV文件名。该函数先获取豆瓣图书的HTML并解析出数据，然后将数据写入CSV文件中。

def crawl_book(url, filename):
    html = yield from fetch_content(url)
    data = list(parse_html(html))
    save_csv(filename, data)
    print(f'Crawled {url} successfully.')

接下来，我们在主程序中使用协程爬取豆瓣图书数据。

首先，定义待爬取的图书列表和CSV文件名。

urls = [f'https://book.douban.com/tag/编程?start={i*20}' for i in range(5)]
filename = 'books.csv'

定义一个协程任务调度函数，使用 asyncio 库实现。

import asyncio


async def task(urls, filename):
    tasks = [crawl_book(url, filename) for url in urls]
    await asyncio.gather(*tasks)

使用以下代码启动协程任务调度函数。

start_time = time.time()
loop = asyncio.get_event_loop()
loop.run_until_complete(task(urls, filename))
elapsed_time = time.time() - start_time
print(f'Total time elapsed: {elapsed_time:.2f}s')

运行程序后，完成爬取任务，并将数据保存在 books.csv 文件中。运行时间大幅度降低，可以看出协程的优势。

示例二：实现异步UDP服务器

以下示例使用Python3.7+提供的新语法 async 和 await 实现异步UDP服务器。

import asyncio


async def handle_datagram(remote_addr, message):
    # 模拟消息处理的耗时过程
    await asyncio.sleep(0.1)
    print(f'Received a message {message.decode()} from {remote_addr[0]}:{remote_addr[1]}')


async def listen_udp_server(host, port):
    # 创建UDP服务器
    server = await asyncio.create_datagram_endpoint(
        protocol_factory=asyncio.DatagramProtocol,
        local_addr=(host, port)
    )

    print(f'Server started and listening on {host}:{port}.')

    # 监听客户端发送的UDP数据包
    async for datagram, remote_addr in server[0]:
        asyncio.create_task(handle_datagram(remote_addr, datagram))


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(listen_udp_server('127.0.0.1', 8888))
    except KeyboardInterrupt:
        pass
    finally:
        loop.close()

运行以上代码后，会启动一个监听本地8888端口的异步UDP服务器，并接收客户端发送的消息。当收到消息后，调用 handle_datagram 函数处理，模拟一个耗时的操作。

总结

协程是一种效率较高的异步编程方式，可以充分发挥计算机的性能，提高程序的并发性和性能。Python提供了多种实现协程的方式，如使用生成器实现、使用asyncio库实现、使用gevent库实现等。

在使用协程时，需要合理使用 yield 和 send() 函数，避免死锁和资源泄漏等问题。同时也需要注意调整好协程任务的数量和优先级。

最后，协程的编写需要一定的技巧和经验，希望本文的介绍能够对读者有所帮助。

本站文章如无特殊说明，均为本站原创，如若转载，请注明出处：Python协程原理全面分析 - Python技术站

Python协程原理全面分析