【Python爬虫实战】深入理解Python异步编程：从协程基础到高效爬虫实现

易辰君

发布于 2024-11-07 22:25:36

610

发布于 2024-11-07 22:25:36

文章被收录于专栏：CSDN

前言

随着网络和数据的迅速发展，越来越多的场景需要高效处理大量请求和数据。传统的同步编程模式在处理I/O密集型任务时会浪费大量等待时间，而Python的异步编程技术提供了一种更高效的方式。本文从Python异步编程的基础概念出发，深入讲解协程、asyncio库及其核心功能。通过详细的代码示例与解释，我们将逐步探索异步编程的应用场景

一、异步

在Python中，异步编程是一种并发编程方法，允许程序在处理耗时任务时不必等待任务完成，而是继续执行其他代码。这样可以提升程序的效率和响应速度，特别适合处理I/O密集型任务（如网络请求、文件读写等）。

（一）核心概念

（1）事件循环：异步编程的核心是事件循环（Event Loop），它管理任务的调度。事件循环会不断地检查是否有任务完成或需要开始新任务，从而实现任务的非阻塞执行。

（2）协程（Coroutine）：协程是异步任务的基本单元，是一个可以被挂起并在稍后继续执行的函数。Python 通过 async def 定义协程函数，协程内部可以用 await 来暂停并等待其他协程的结果。

（3）async 和 await 关键字：

async 用于定义一个协程函数，例如 async def my_function()。
await 用于暂停协程的执行并等待其他协程完成。例如，await some_async_task() 会暂停当前协程，直到 some_async_task() 完成后再继续执行。

（4）asyncio 库： Python 的标准库 asyncio 提供了异步编程的核心功能，包含事件循环、任务管理、以及异步 I/O 操作等工具，帮助处理并发任务。

（二）应用场景

异步编程非常适合处理以下场景：

网络请求（如 HTTP 请求、数据库查询等）
文件读写操作
大量并发任务（如网络爬虫、数据采集）

示例：

import asyncio

async def task(name, delay):
    print(f"Task {name} started")
    await asyncio.sleep(delay)  # 模拟一个异步I/O操作
    print(f"Task {name} completed")

async def main():
    # 并发执行多个任务
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )

# 运行主协程
asyncio.run(main())

在这个例子中，task 是一个协程，使用 await asyncio.sleep(delay) 来模拟一个耗时任务。在 main 函数中，asyncio.gather 可以并发地执行多个 task，而不需要等待其中一个任务完成才执行下一个。

（三）优缺点

优点：相比传统同步方法，异步编程在 I/O 密集型任务上更高效、响应更快。
缺点：异步代码的逻辑较难理解和调试，并且在 CPU 密集型任务上并不一定有优势。

二、协程异步实现方法

在Python中，使用协程实现异步的主要方法是通过 async 和 await 关键字以及 asyncio 库来管理协程和事件循环。这使得我们能够编写出非阻塞的代码，有效地进行异步任务调度。下面是几种常用的协程异步实现方法：

（一）基本的协程定义与运行

import asyncio

async def my_task():
    print("Task started")
    await asyncio.sleep(1)  # 模拟异步操作
    print("Task completed")

# 运行协程
asyncio.run(my_task())

在这个例子中：

my_task 是一个协程，使用 async def 定义。
await asyncio.sleep(1) 模拟一个异步操作。在实际应用中，这可以替换为其他 I/O 操作，比如网络请求。

（二）并发执行多个协程

可以使用 asyncio.gather 并发运行多个协程，将它们一起调度，以便程序在等待一个任务时可以继续执行其他任务：

async def task(name, delay):
    print(f"Task {name} started")
    await asyncio.sleep(delay)
    print(f"Task {name} completed")

async def main():
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )

asyncio.run(main())

在这里，asyncio.gather 接收多个协程，创建一个任务组，多个任务会并发执行，节省时间。

（三）创建与管理任务

asyncio.create_task 将协程封装成任务并立即调度，而不需要等待所有任务完成：

async def task(name):
    print(f"Task {name} started")
    await asyncio.sleep(1)
    print(f"Task {name} completed")

async def main():
    task1 = asyncio.create_task(task("A"))
    task2 = asyncio.create_task(task("B"))
    
    # 可以在这里添加其他逻辑
    await task1
    await task2

asyncio.run(main())

通过 asyncio.create_task，可以先开始任务，然后在稍后使用 await 等待其完成。这样可以在需要时调度多个任务，并在合适的位置等待结果。

（四）限制并发数

在某些场景中需要限制并发数，可以使用 asyncio.Semaphore 控制：

import asyncio

semaphore = asyncio.Semaphore(2)

async def limited_task(name, delay):
    async with semaphore:
        print(f"Task {name} started")
        await asyncio.sleep(delay)
        print(f"Task {name} completed")

async def main():
    await asyncio.gather(
        limited_task("A", 2),
        limited_task("B", 1),
        limited_task("C", 3),
        limited_task("D", 1)
    )

asyncio.run(main())

在这个例子中，asyncio.Semaphore(2) 设置同时最多只能有两个任务在执行，其他任务必须等待。

（五）超时控制

可以使用 asyncio.wait_for 来限制某个任务的最长等待时间：

async def my_task():
    await asyncio.sleep(3)
    print("Task completed")

async def main():
    try:
        await asyncio.wait_for(my_task(), timeout=2)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

（六）队列管理

asyncio.Queue 是实现生产者-消费者模式的常用方式，可以让多个协程通过队列共享数据：

async def producer(queue):
    for i in range(5):
        await asyncio.sleep(1)
        await queue.put(f"Item {i}")
        print(f"Produced Item {i}")

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    producer_task = asyncio.create_task(producer(queue))
    consumer_task = asyncio.create_task(consumer(queue))
    
    await producer_task
    await queue.put(None)  # 终止消费者
    await consumer_task

asyncio.run(main())

在这里，producer 生产数据并放入队列，consumer 从队列中消费数据。队列的使用可以很方便地控制协程之间的数据传递和同步。

三、同步和异步的对比

同步和异步是两种处理任务的不同方式。它们在任务的执行和等待机制上有显著的区别，适合不同的应用场景。以下是它们的详细对比：

（一）执行方式

同步：任务按照顺序逐个执行，当前任务完成后才能执行下一个任务。如果一个任务正在执行，其他任务必须等待。
异步：任务可以在不等待其他任务完成的情况下启动，任务之间的执行不严格依赖顺序，多个任务可以同时进行（在I/O操作上，异步非常有效）。

（二）阻塞与非阻塞

同步：同步方式是阻塞的，任务在执行期间会阻塞代码的后续执行，直到任务完成才会继续执行下一步。
异步：异步方式是非阻塞的，一个任务开始后可以立即开始执行其他任务，不必等待前一个任务完成。

（三）性能和效率

同步：在I/O操作频繁的程序中，同步会导致时间浪费，因为程序在等待I/O操作完成时处于空闲状态。适用于计算密集型或对顺序要求严格的场景。
异步：通过避免等待，提高了效率和响应速度。特别适用于I/O密集型操作（如网络请求、文件读写等），异步允许程序在等待I/O操作完成时继续处理其他任务。

（四）代码复杂性

同步：代码结构相对简单，因为任务是顺序执行的，不涉及任务切换和状态管理，容易理解和调试。
异步：代码相对复杂，尤其是在大型项目中，由于任务非顺序执行，涉及事件循环、回调或await/async等机制，代码逻辑可能较难理解和维护。

（五）应用场景

同步：适用于对任务顺序有严格要求的场景，例如：数据处理、算法计算、特定顺序要求的逻辑。
异步：适合需要处理大量I/O操作、需要高并发支持的场景，例如：网络爬虫、聊天应用、实时数据流处理等。

（六）示例对比

同步示例

import time

def task(name):
    print(f"Starting task {name}")
    time.sleep(2)  # 模拟阻塞操作
    print(f"Task {name} completed")

# 顺序执行
task("A")
task("B")
task("C")

输出：

Starting task A
Task A completed
Starting task B
Task B completed
Starting task C
Task C completed

在同步示例中，task 函数按顺序执行，每个任务完成后才开始下一个。

异步示例

import asyncio

async def task(name):
    print(f"Starting task {name}")
    await asyncio.sleep(2)  # 模拟非阻塞操作
    print(f"Task {name} completed")

async def main():
    await asyncio.gather(task("A"), task("B"), task("C"))

asyncio.run(main())

输出：

Starting task A
Starting task B
Starting task C
Task A completed
Task B completed
Task C completed

在异步示例中，任务A、B、C几乎同时开始，await asyncio.sleep(2)不会阻塞其他任务，任务可以并发执行，最终加快了整体运行速度。

对比总结

特性	同步	异步
执行方式	顺序执行	并发执行
阻塞	阻塞	非阻塞
效率	I/O 密集型性能低	I/O 密集型性能高
代码复杂度	简单	较复杂
适用场景	计算密集型任务	I/O 密集型、高并发任务

在实际应用中，同步和异步各有优缺点。选择同步或异步，主要取决于应用场景、任务需求和性能要求。

四、异步爬虫

异步爬虫是一种使用异步编程方法实现的网络爬虫，它能够在不等待网页响应的情况下，同时发送多个请求并处理返回的数据。这种方式特别适用于需要抓取大量网页内容的场景，因为它可以显著提升爬虫的效率。

在Python中，异步爬虫通常使用 asyncio 和 aiohttp 两个库来实现：

asyncio：提供异步编程的核心框架，包括事件循环、协程和任务管理。
aiohttp：一个异步HTTP库，支持异步发送请求和获取响应，非常适合构建异步爬虫。

（一）异步爬虫的优点

高并发性：可以同时发送大量请求，而不必等待每个请求完成再发送下一个。
资源利用率高：在等待服务器响应时可以处理其他任务，减少了等待时间。
适合I/O密集型任务：异步爬虫特别适用于抓取数据量大、网络请求多的任务场景。

（二）实现异步爬虫的基本步骤

以下是一个使用 asyncio 和 aiohttp 构建异步爬虫的示例，展示如何同时请求多个网页并处理响应。

示例：

import asyncio
import aiohttp

# 定义一个异步请求函数
async def fetch(session, url):
    try:
        async with session.get(url) as response:
            print(f"Fetching {url}")
            html = await response.text()  # 获取网页的 HTML 内容
            print(f"Completed {url}")
            return html
    except Exception as e:
        print(f"Error fetching {url}: {e}")

# 主爬虫函数
async def main(urls):
    # 创建一个aiohttp会话
    async with aiohttp.ClientSession() as session:
        # 使用 asyncio.gather 并发请求多个 URL
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# 运行爬虫
urls = [
    "https://example.com",
    "https://example.org",
    "https://example.net"
]

# 启动事件循环
asyncio.run(main(urls))

代码解析

fetch(session, url)：这是异步请求函数，使用 session.get(url) 发送异步 HTTP 请求。async with 确保会话资源能够正确释放。
asyncio.gather(*tasks)：将所有 fetch 请求作为任务传入 asyncio.gather，这样可以并发地执行这些任务，而不需要等待每个任务顺序完成。
asyncio.run(main(urls))：启动事件循环并运行 main 函数，main 中会创建多个并发任务并等待它们的完成。

（三）控制并发数量

在实际应用中，为了防止服务器拒绝请求，可以使用 asyncio.Semaphore 来限制并发请求数量。例如，限制并发请求数为5：

async def fetch(semaphore, session, url):
    async with semaphore:  # 使用信号量限制并发
        try:
            async with session.get(url) as response:
                print(f"Fetching {url}")
                html = await response.text()
                print(f"Completed {url}")
                return html
        except Exception as e:
            print(f"Error fetching {url}: {e}")

async def main(urls):
    semaphore = asyncio.Semaphore(5)  # 限制为5个并发请求
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(semaphore, session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

（四）应用场景

数据采集：高效采集电商网站、新闻网站等的商品或内容信息。
实时数据爬取：抓取实时更新的内容，如股票数据、天气数据等。
大规模网页抓取：异步爬虫非常适合抓取大量网页内容，因为它能在不等待单个网页响应的情况下发起多个请求。

（五）注意事项

速率控制：可以加入请求延迟或限制并发请求数量，以防止被目标网站封禁。
异常处理：要处理网络超时、连接错误等异常，保证爬虫在出现错误时不会中断。
机器人检测：一些网站会检测并阻止大量并发请求，需要考虑反爬策略（如伪装请求头、模拟浏览器等）。

五、aiomysql的使用

aiomysql 是一个支持 Python 异步编程的 MySQL 数据库库，基于 asyncio 和 PyMySQL 构建。它可以让开发者在异步框架中执行数据库操作，适合需要同时处理大量数据库请求的高并发应用，如爬虫数据存储、Web 服务等。

（一）特点

异步支持：基于 asyncio 的异步支持，不会因为等待数据库响应而阻塞其他任务。
高效连接池：提供了内置的数据库连接池，减少每次查询前创建新连接的开销。
灵活的事务处理：支持事务和多种数据库操作，适合复杂的数据库事务操作。

（二）安装

在使用前，需要安装 aiomysql。可以通过以下命令进行安装：

pip install aiomysql

（三）使用示例

以下是一个简单的 aiomysql 示例，包括如何创建连接、执行查询、插入数据和使用连接池。

（1）创建连接并执行查询

import asyncio
import aiomysql

async def main():
    # 创建连接
    conn = await aiomysql.connect(
        host='localhost', port=3306,
        user='root', password='password',
        db='test_db'
    )

    async with conn.cursor() as cur:
        # 执行查询
        await cur.execute("SELECT * FROM example_table")
        result = await cur.fetchall()  # 获取所有查询结果
        print(result)

    conn.close()  # 关闭连接

# 运行异步主函数
asyncio.run(main())

（2）使用连接池

使用连接池可以避免频繁创建和关闭连接，适合大量并发请求的场景。

import asyncio
import aiomysql

async def query_with_pool(pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM example_table")
            result = await cur.fetchall()
            print(result)

async def main():
    # 创建连接池
    pool = await aiomysql.create_pool(
        host='localhost', port=3306,
        user='root', password='password',
        db='test_db', maxsize=10
    )

    # 执行查询
    await query_with_pool(pool)
    pool.close()
    await pool.wait_closed()  # 关闭连接池

asyncio.run(main())

（3）插入数据示例

import asyncio
import aiomysql

async def insert_data():
    conn = await aiomysql.connect(
        host='localhost', port=3306,
        user='root', password='password',
        db='test_db'
    )
    async with conn.cursor() as cur:
        await cur.execute(
            "INSERT INTO example_table (name, age) VALUES (%s, %s)",
            ('Alice', 25)
        )
        await conn.commit()  # 提交事务
    conn.close()

asyncio.run(insert_data())

（4）异步事务处理

在数据需要严格一致性时，可以使用事务：

async def transaction_example():
    conn = await aiomysql.connect(
        host='localhost', port=3306,
        user='root', password='password',
        db='test_db'
    )

    async with conn.cursor() as cur:
        try:
            await cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
            await cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
            await conn.commit()  # 提交事务
        except Exception as e:
            await conn.rollback()  # 回滚事务
            print("Transaction failed:", e)
        finally:
            conn.close()

asyncio.run(transaction_example())

（5）常见注意事项

错误处理：在异步环境中，尽量在执行数据库操作时捕获异常，避免由于未处理的异常导致协程退出。
事务一致性：在批量插入、转账等操作中，建议使用事务保证数据一致性。
连接池管理：使用 aiomysql 的连接池，尤其在高并发场景中，能够显著提高数据库访问的性能。

六、总结

Python异步编程通过非阻塞的事件循环实现了并发任务调度，特别适合处理I/O密集型任务，如网络请求、文件读写等。在本文中，我们探讨了异步编程的核心概念与实现方式，包括协程、事件循环、并发控制等。基于这些技术，还展示了如何利用asyncio和aiohttp构建高效的异步爬虫。掌握这些异步编程方法，不仅能大幅提升代码执行效率，还为处理大规模数据和并发任务提供了强有力的工具。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2024-10-30，如有侵权请联系 cloudcommunity@tencent.com 删除

基础

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度