Python 生成器和协程

数据处理领域，高效处理大型数据集是一个常见的挑战，尤其是在资源有限的情况下。传统方法往往会导致内存使用率高、性能低下，给数据分析和应用开发带来瓶颈。Python 以其简单性和强大功能而闻名，它为这些挑战提供了两种优雅的解决方案：生成器和协程。这些功能不仅优化了内存使用，还提高了数据处理操作的效率。

2. 了解 Python 生成器

2.1. 什么是生成器？

Python 中的生成器是用于创建迭代器的简单而强大的工具。它们像常规函数一样编写， yield 但使用语句返回数据。这种机制允许生成器随着时间的推移生成一系列值，而不是一次计算和存储它们。例如，生成某个范围内的数字的生成器函数使用的内存比创建这些数字列表的等效函数少得多。这是因为它动态生成每个数字，并且仅在内存中维护当前状态。

例：

def count_up_to(max): count = 1 while count <= max: yield count count += 1counter = count_up_to(5)for number in counter: print(number)

此生成器 count_up_to ，生成最多指定最大值的数字。

输出：

12345

2.2. 生成器的工作原理

生成器通过实现迭代器协议（一组方法（ __iter__ 和 __next__ ）在后台工作，允许在for循环中迭代对象。与返回单个值并退出的常规函数不同，生成器按顺序生成多个值，在每个 yield 值之后暂停，并在下一次调用时从那里恢复。考虑一个现实世界的类比：发电机就像工厂里的传送带，根据需要一次生产一件物品，而不是提前储存。

例：

def fibonacci(): a, b = 0, 1 while True: yield a a, b = b, a + bfib = fibonacci()for _ in range(10): print(next(fib))

此斐波那契生成器演示了生成器如何在迭代中保持状态。

输出：

0112358132134

2.3. 使用生成器的好处

2.3.1. 内存效率

在内存使用方面，生成器比传统的基于集合的方法（如列表）具有明显的优势。例如，使用生成器逐行处理大型文件只需要当前行的内存，这与将整个文件读入列表不同，后者消耗的内存与文件大小成正比。这使得生成器非常适合在 Python 中处理内存效率高的数据。

例：

# Generator for large datadef large_file_reader(file_name): for row in open(file_name, 'r'): yield row# Reading a large filefor line in large_file_reader('large_file.txt'): process(line)

此示例逐行读取大文件，而不将其完全加载到内存中。

2.3.2. 懒惰和性能

生成器是惰性的，这意味着它们仅在需要时动态生成值。这种延迟评估会导致性能优化，尤其是在不需要所有生成值的情况下。它允许对大型数据集进行高效的循环和处理，而无需将整个数据集加载到内存中的开销。

例：

# Infinite sequence generatordef even_numbers():    n = 0    while True:        yield n        n += 2evens = even_numbers()for _ in range(5):    print(next(evens))

生成无限的偶数序列，但仅在需要时生成。

输出：

02468

2.4. 发电机的实际应用

2.4.1. 使用大型数据集

在数据处理和分析中，生成器对于处理大型数据集非常宝贵。它们能够以最小的内存占用处理太大而无法放入内存的数据流或文件，例如日志文件分析或大型 CSV 处理。

例：

# Generator for large CSV processingdef csv_reader(file_name):    for row in open(file_name, 'r'):        yield row.split(',')# Processing large CSVfor data in csv_reader('large_dataset.csv'):    analyze(data)

逐行处理大型 CSV 文件。

2.4.2. 简化数据管道

生成器还可用于创建高效的数据管道。例如，在涉及读取数据、转换数据然后写出数据的数据处理管道中，每个步骤都可以是一个生成器，将数据从一个步骤传递到下一个步骤，而无需将所有数据加载到内存中。

例：

# Data pipeline with generatorsdef read_data(file_name): for row in open(file_name, 'r'): yield rowdef filter_data(rows): for row in rows: if condition(row): yield row# Using the pipelinefiltered_data = filter_data(read_data('data.txt'))for data in filtered_data: process(data)

这演示了读取和筛选数据的管道。

注意：单击此处了解有关 Python 中生成器的更多信息。

3. 协程简介

3.1. 什么是协程？

协程是 Python 中允许异步编程的一项功能。与生成器不同，生成器使用 yield 生成一系列值，协程旨在处理异步任务。它们用于协作多任务处理，其中函数可以暂停其执行，直到满足某些条件或完成操作，而不会阻塞整个程序流。

3.2. 协程的工作原理

Python 中的协程是使用语法来定义异步函数和关键字来实现的，该 async def await 语法用于暂停协程的执行，直到等待的操作完成。这种非阻塞行为在异步编程中是必不可少的，它由事件循环管理，例如 asyncio 模块提供的事件循环。事件循环运行异步任务和回调，管理它们之间的通信，并处理 I/O 事件。

例：

import asyncioasync def task(name, delay):    print(f'Task {name} starting with delay {delay}')    await asyncio.sleep(delay)    print(f'Task {name} completed after {delay} seconds')async def main():    # Schedule multiple tasks concurrently    await asyncio.gather(        task('A', 2),  # Task A will take 2 seconds to complete        task('B', 3),  # Task B will take 3 seconds to complete        task('C', 1)   # Task C will take 1 second to complete    )    print('All tasks completed')# Run the main coroutineasyncio.run(main())

在此代码中：

1. 协程定义：我们定义一个异步函数 task ，它表示需要一定时间才能完成的任务。该 async def 语法用于定义此协程。
2. 异步等待：在每个任务中，我们使用 await asyncio.sleep(delay) .这模拟了等待操作（如 I/O）完成的任务。这里的关键点是，当任务处于“休眠状态”时，它不会阻止其他任务的执行。
3. 并发执行：在协程中 main ， asyncio.gather 用于并发运行多个实例 task 。这演示了协作式多任务处理，即使在它们之后开始， Task C 也会在它们之前 Task A 完成 Task B 。
4. 运行协程： asyncio.run(main()) 启动事件循环，用于管理协程的执行。

输出：

Task A starting with delay 2Task B starting with delay 3Task C starting with delay 1Task C completed after 1 secondsTask A completed after 2 secondsTask B completed after 3 secondsAll tasks completed

3.3. 利用协程执行异步任务

3.3.1. Python 中的异步编程

使用协程的异步编程对于 I/O 受限和高延迟活动特别有效。与传统的线程不同，使用协程的异步编程允许在单个线程中并发管理多个任务，从而减少开销和复杂性。这在处理多个网络连接或执行需要等待外部资源的操作等方案中特别有用。

例：

import asyncioimport aiohttpasync def fetch(url, session):    async with session.get(url) as response:        return await response.text()async def main():    urls = [        'http://google.com',        'http://yahoo.com',        'http://',    ]    async with aiohttp.ClientSession() as session:        tasks = [fetch(url, session) for url in urls]        responses = await asyncio.gather(*tasks)        for response in responses:            print(response[:100])  # Print first 100 characters of each responsedef run_asyncio_coroutine():    try:        # Get the current event loop, but don't close it afterwards        loop = asyncio.get_event_loop()    except RuntimeError as e:        # If no current event loop, create a new one        loop = asyncio.new_event_loop()        asyncio.set_event_loop(loop)        # Run the coroutine    loop.run_until_complete(main())# Execute the functionrun_asyncio_coroutine()

代码说明：

• async def：用于定义异步函数（或“协程”）。定义的 async def 函数可用于 await 调用其他异步函数。
• 等待：
  
  暂停当前协程的执行，等待异步操作完成，然后在操作完成后恢复协程。
• asayankio.sleep（）：
  
  模拟给定秒数的延迟（或睡眠）的异步函数。它将控制权交还给事件循环，允许其他操作在睡眠期间运行。
• asyncio.gather（）：
  
  将多个异步任务（协程）调度为并发运行的函数。它返回一个聚合任务结果的单个未来。
• 异步.run（）：
  
  用于执行给定协程并返回其结果的函数。它用于运行异步程序的主协程。当另一个 asyncio 事件循环在同一线程中运行时，无法调用它。
• asyncio.get_event_loop（）：
  
  获取当前 asyncio 事件循环。事件循环是 asyncio 异步执行的核心，用于处理异步任务和回调的执行。
• asyncio.new_event_loop（）：
  
  创建新的事件循环。在当前线程可能没有事件循环或现有线程已关闭的情况下很有用。
• aiohttp。客户端会话（）：
  
  aiohttp 库提供的用于管理 HTTP 连接的类。它用作上下文管理器（ async with) 以确保正确的资源清理（如关闭会话）。
• session.get（网址）：
  
  用于对指定 URL 执行 HTTP GET 请求的 aiohttp.ClientSession 异步方法。

3.3.2. 异步数据库操作

协程可用于执行异步数据库操作。当必须同时处理多个数据库查询或事务时，这特别有用。

在此示例中，我们将使用 aiomysql ，一个用于 MySQL 的异步库。首先，安装它：

pip install aiomysql

下面是与 asyncio 一起使用 aiomysql 的基本示例：

import asyncioimport aiomysqlasync def fetch_data(pool):    async with pool.acquire() as conn:        async with conn.cursor() as cur:            await cur.execute('SELECT * FROM my_table')            result = await cur.fetchall()            return resultasync def main():    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,                                      user='root', password='password',                                      db='my_database', loop=asyncio.get_event_loop())        data = await fetch_data(pool)    for row in data:        print(row)  # Process each row       pool.close()    await pool.wait_closed()asyncio.run(main())

在这些示例中， asyncio.run() 用于运行主协程， async with 处理异步上下文管理（用于 HTTP 会话或数据库连接），并 await 用于等待异步操作（如网络请求或数据库查询）完成。这些是 Python 中协程的典型用例，展示了它们在处理异步 I/O 操作方面的有效性。