IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen python案例 249

本文目录导读：

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

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

我来详细介绍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())

注意事项

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

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

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python脚本如何批量发起协程请求

下一篇Python脚本如何优化协程IO阻塞问题

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝