Python脚本如何搭配协程网络请求

wen python案例 2026-06-27 29

本文目录导读：

Python脚本如何搭配协程网络请求

基础环境准备
使用 aiohttp 实现协程网络请求
使用 httpx 库（支持同步和异步）
实际应用：批量下载图片
高级用法：带重试机制的协程请求
性能监控和限流
实战：网页爬虫示例
注意事项

我来详细介绍Python中协程配合网络请求的几种常见方式。

基础环境准备

# 安装必要的库
# pip install aiohttp httpx

使用 aiohttp 实现协程网络请求

基础用法

import asyncio
import aiohttp
async def fetch_url(session, url):
    """异步获取单个URL"""
    try:
        async with session.get(url, timeout=10) as response:
            return await response.text()
    except Exception as e:
        return f"Error: {e}"
async def main():
    urls = [
        'https://api.github.com',
        'https://httpbin.org/get',
        'https://jsonplaceholder.typicode.com/posts/1'
    ]
    async with aiohttp.ClientSession() as session:
        # 并发请求多个URL
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, result in zip(urls, results):
            print(f"URL: {url}")
            print(f"Response length: {len(result)} chars")
            print("-" * 50)
# 运行
asyncio.run(main())

带参数和headers的请求

import asyncio
import aiohttp
import json
async def fetch_with_params():
    headers = {
        'User-Agent': 'Mozilla/5.0',
        'Authorization': 'Bearer your_token'
    }
    params = {
        'page': 1,
        'limit': 10
    }
    async with aiohttp.ClientSession(headers=headers) as session:
        # GET请求带参数
        async with session.get('https://api.example.com/data', 
                              params=params) as resp:
            data = await resp.json()
            return data
        # POST请求
        # async with session.post('https://api.example.com/create',
        #                        json={'key': 'value'}) as resp:
        #     return await resp.json()

使用 httpx 库（支持同步和异步）

import asyncio
import httpx
async def httpx_example():
    async with httpx.AsyncClient() as client:
        # 并发请求
        async with client as c:
            tasks = [
                c.get('https://api.github.com'),
                c.get('https://httpbin.org/get'),
                c.get('https://jsonplaceholder.typicode.com/posts')
            ]
            responses = await asyncio.gather(*tasks)
            for response in responses:
                print(f"Status: {response.status_code}")
                print(f"Content-Length: {len(response.text)} bytes")

实际应用：批量下载图片

import asyncio
import aiohttp
import aiofiles
from pathlib import Path
class AsyncDownloader:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.download_dir = Path("downloads")
        self.download_dir.mkdir(exist_ok=True)
    async def download_file(self, session, url, filename):
        async with self.semaphore:  # 控制并发数
            try:
                async with session.get(url) as response:
                    if response.status == 200:
                        content = await response.read()
                        # 异步写入文件
                        async with aiofiles.open(
                            self.download_dir / filename, 'wb'
                        ) as f:
                            await f.write(content)
                        return f"Downloaded: {filename}"
                    else:
                        return f"Failed: {url} - Status: {response.status}"
            except Exception as e:
                return f"Error: {url} - {e}"
    async def batch_download(self, urls_with_names):
        async with aiohttp.ClientSession() as session:
            tasks = []
            for url, filename in urls_with_names:
                task = self.download_file(session, url, filename)
                tasks.append(task)
            results = await asyncio.gather(*tasks)
            return results
# 使用示例
async def main():
    downloader = AsyncDownloader(max_concurrent=5)
    urls = [
        ("https://httpbin.org/image/jpeg", "image1.jpg"),
        ("https://httpbin.org/image/png", "image2.png"),
        # 更多URL...
    ]
    results = await downloader.batch_download(urls)
    for result in results:
        print(result)
# asyncio.run(main())

高级用法：带重试机制的协程请求

import asyncio
import aiohttp
from typing import Optional
from tenacity import retry, stop_after_attempt, wait_exponential
class RobustAsyncClient:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    async def fetch_with_retry(self, session, url, **kwargs):
        async with session.get(url, **kwargs) as response:
            response.raise_for_status()
            return await response.json()
    async def fetch_all(self, urls):
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.fetch_with_retry(session, url) 
                for url in urls
            ]
            return await asyncio.gather(*tasks, return_exceptions=True)
# 使用
async def main():
    client = RobustAsyncClient()
    urls = [
        'https://api.github.com',
        'https://invalid-url.example.com',  # 这个会失败并重试
        'https://httpbin.org/delay/2'       # 延迟响应
    ]
    results = await client.fetch_all(urls)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"URL {i} failed: {result}")
        else:
            print(f"URL {i} succeeded: {len(str(result))} chars")

性能监控和限流

import asyncio
import aiohttp
import time
from collections import deque
class RateLimiter:
    def __init__(self, max_calls=10, period=1.0):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
    async def acquire(self):
        while True:
            now = time.time()
            # 清理过期记录
            while self.calls and self.calls[0] <= now - self.period:
                self.calls.popleft()
            if len(self.calls) < self.max_calls:
                self.calls.append(now)
                return
            else:
                # 等待直到有可用配额
                sleep_time = self.calls[0] + self.period - now
                await asyncio.sleep(sleep_time)
async def rate_limited_fetch():
    limiter = RateLimiter(max_calls=5, period=1.0)  # 每秒5个请求
    async with aiohttp.ClientSession() as session:
        for i in range(20):
            await limiter.acquire()
            async with session.get('https://httpbin.org/delay/0.1') as resp:
                print(f"Request {i+1}: Status {resp.status}")

实战：网页爬虫示例

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AsyncWebScraper:
    def __init__(self, concurrent_limit=10):
        self.semaphore = asyncio.Semaphore(concurrent_limit)
    async def fetch_page(self, session, url):
        """获取单个页面"""
        async with self.semaphore:
            try:
                async with session.get(url, timeout=30) as response:
                    return await response.text()
            except Exception as e:
                logger.error(f"Failed to fetch {url}: {e}")
                return None
    async def parse_page(self, html, url):
        """解析页面内容"""
        if not html:
            return None
        soup = BeautifulSoup(html, 'html.parser')
        title = soup.title.string if soup.title else "No title"
        links = [a.get('href') for a in soup.find_all('a', href=True)]
        return {
            'url': url,
            'title': title,
            'links': links[:10]  # 只取前10个链接
        }
    async def crawl(self, urls):
        """爬取多个页面"""
        async with aiohttp.ClientSession() as session:
            # 并发获取页面
            fetch_tasks = [self.fetch_page(session, url) for url in urls]
            htmls = await asyncio.gather(*fetch_tasks)
            # 解析页面
            parse_tasks = [
                self.parse_page(html, url) 
                for html, url in zip(htmls, urls)
            ]
            results = await asyncio.gather(*parse_tasks)
            return [r for r in results if r is not None]
# 使用
async def main():
    scraper = AsyncWebScraper(concurrent_limit=5)
    urls = [
        'https://python.org',
        'https://github.com',
        'https://stackoverflow.com'
    ]
    results = await scraper.crawl(urls)
    for result in results:
        print(f"URL: {result['url']}")
        print(f"Title: {result['title']}")
        print(f"Links found: {len(result['links'])}")
        print("-" * 50)
# asyncio.run(main())

注意事项

错误处理：协程网络请求一定要做好异常处理
连接池管理：复用 ClientSession 而不是每次创建
超时设置：总超时和请求级别超时都要设置
资源限制：使用信号量控制并发数
内存管理：大响应使用流式处理，不要一次性加载到内存

这些示例涵盖了协程网络请求的常见场景,你可以根据实际需求选择合适的方案。