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

Python异步请求案例如何异步调用

wen python案例 26

本文目录导读：

Python异步请求案例如何异步调用

  1. 基本异步请求 (使用 aiohttp)
  2. 使用 httpx (支持同步和异步)
  3. 带并发限制的异步请求
  4. 带重试机制的异步请求
  5. 实际应用：批量下载图片

我来为你介绍Python异步请求的几种常见方式,并给出实际案例。

基本异步请求 (使用 aiohttp)

import asyncio
import aiohttp
import time
async def fetch_url(session, url):
    """异步获取URL内容"""
    try:
        async with session.get(url) as response:
            # 等待响应完成
            content = await response.text()
            return {
                'url': url,
                'status': response.status,
                'length': len(content)
            }
    except Exception as e:
        return {
            'url': url,
            'error': str(e)
        }
async def main():
    """主异步函数"""
    urls = [
        'http://httpbin.org/delay/1',
        'http://httpbin.org/delay/2',
        'http://httpbin.org/delay/3',
        'http://httpbin.org/get',
        'http://httpbin.org/ip'
    ]
    # 创建连接会话
    async with aiohttp.ClientSession() as session:
        # 创建所有任务
        tasks = [fetch_url(session, url) for url in urls]
        # 并发执行所有任务
        results = await asyncio.gather(*tasks)
        # 输出结果
        for result in results:
            print(f"URL: {result['url']}")
            if 'error' in result:
                print(f"错误: {result['error']}")
            else:
                print(f"状态码: {result['status']}, 内容长度: {result['length']}")
            print("-" * 50)
# 运行异步函数
start_time = time.time()
asyncio.run(main())
print(f"总耗时: {time.time() - start_time:.2f}秒")

使用 httpx (支持同步和异步)

import asyncio
import httpx
import time
async def fetch_url_httpx(client, url):
    """使用httpx异步获取"""
    try:
        response = await client.get(url)
        return {
            'url': url,
            'status': response.status_code,
            'content': response.text[:100]  # 只取前100字符
        }
    except Exception as e:
        return {'url': url, 'error': str(e)}
async def main_httpx():
    """httpx主函数"""
    urls = [
        'http://httpbin.org/get',
        'http://httpbin.org/ip',
        'http://httpbin.org/user-agent'
    ]
    async with httpx.AsyncClient() as client:
        tasks = [fetch_url_httpx(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(f"URL: {result['url']}")
            if 'error' in result:
                print(f"错误: {result['error']}")
            else:
                print(f"状态码: {result['status']}")
            print("-" * 50)
start_time = time.time()
asyncio.run(main_httpx())
print(f"总耗时: {time.time() - start_time:.2f}秒")

带并发限制的异步请求

import asyncio
import aiohttp
from asyncio import Semaphore
class AsyncRequestLimiter:
    """带并发限制的异步请求器"""
    def __init__(self, max_concurrent=5):
        self.semaphore = Semaphore(max_concurrent)
    async def fetch_with_limit(self, session, url):
        """带限制的请求"""
        async with self.semaphore:  # 控制并发数
            return await self._fetch(session, url)
    async def _fetch(self, session, url):
        """实际请求方法"""
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
                return {
                    'url': url,
                    'status': response.status,
                    'data': await response.text()
                }
        except Exception as e:
            return {'url': url, 'error': str(e)}
async def main_with_limit():
    """带并发限制的主函数"""
    # 生成100个测试URL
    urls = [f'http://httpbin.org/delay/1' for _ in range(20)]
    limiter = AsyncRequestLimiter(max_concurrent=5)
    async with aiohttp.ClientSession() as session:
        tasks = [limiter.fetch_with_limit(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        # 统计结果
        success = sum(1 for r in results if 'error' not in r)
        failed = sum(1 for r in results if 'error' in r)
        print(f"成功: {success}, 失败: {failed}")
start_time = time.time()
asyncio.run(main_with_limit())
print(f"总耗时: {time.time() - start_time:.2f}秒")

带重试机制的异步请求

import asyncio
import aiohttp
from typing import Optional
async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> Optional[dict]:
    """带重试的异步请求"""
    for attempt in range(max_retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
                if response.status == 200:
                    return {
                        'url': url,
                        'status': response.status,
                        'data': await response.json()
                    }
                elif response.status in [429, 503]:  # 限流或服务不可用
                    wait_time = retry_delay * (2 ** attempt)  # 指数退避
                    print(f"请求被限流，等待 {wait_time} 秒后重试...")
                    await asyncio.sleep(wait_time)
                else:
                    return {
                        'url': url,
                        'status': response.status,
                        'error': f"HTTP {response.status}"
                    }
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < max_retries - 1:
                wait_time = retry_delay * (2 ** attempt)
                print(f"请求失败 ({e})，{wait_time} 秒后重试...")
                await asyncio.sleep(wait_time)
            else:
                return {'url': url, 'error': str(e)}
    return {'url': url, 'error': '达到最大重试次数'}
async def main_with_retry():
    """带重试的主函数"""
    urls = [
        'http://httpbin.org/status/500',  # 会触发重试
        'http://httpbin.org/delay/3',     # 正常请求
        'http://httpbin.org/get'          # 正常请求
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_retry(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(f"URL: {result['url']}")
            if 'error' in result:
                print(f"最终错误: {result['error']}")
            else:
                print(f"状态码: {result['status']}")
            print("-" * 50)
asyncio.run(main_with_retry())

实际应用：批量下载图片

import asyncio
import aiohttp
import aiofiles
import os
from pathlib import Path
class ImageDownloader:
    """图片下载器"""
    def __init__(self, save_dir="downloads", max_concurrent=10):
        self.save_dir = Path(save_dir)
        self.save_dir.mkdir(exist_ok=True)
        self.semaphore = asyncio.Semaphore(max_concurrent)
    async def download_image(self, session: aiohttp.ClientSession, url: str, index: int):
        """下载单张图片"""
        async with self.semaphore:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
                    if response.status == 200:
                        # 获取文件名
                        filename = f"image_{index}.jpg"
                        filepath = self.save_dir / filename
                        # 异步写入文件
                        async with aiofiles.open(filepath, 'wb') as f:
                            await f.write(await response.read())
                        return {'url': url, 'status': 'success', 'file': str(filepath)}
                    else:
                        return {'url': url, 'status': 'failed', 'error': f"HTTP {response.status}"}
            except Exception as e:
                return {'url': url, 'status': 'failed', 'error': str(e)}
    async def download_batch(self, urls: list):
        """批量下载图片"""
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.download_image(session, url, i) 
                for i, url in enumerate(urls)
            ]
            results = await asyncio.gather(*tasks)
            # 统计结果
            success = sum(1 for r in results if r['status'] == 'success')
            failed = sum(1 for r in results if r['status'] == 'failed')
            print(f"下载完成: 成功 {success} 张, 失败 {failed} 张")
            return results
# 使用示例
async def download_images_example():
    downloader = ImageDownloader(max_concurrent=5)
    image_urls = [
        "https://picsum.photos/200/300",
        "https://picsum.photos/200/300",
        "https://picsum.photos/200/300"
    ]
    results = await downloader.download_batch(image_urls)
    for result in results:
        if result['status'] == 'success':
            print(f"下载成功: {result['file']}")
        else:
            print(f"下载失败: {result['error']}")
# asyncio.run(download_images_example())

  1. 异步库选择：

    • aiohttp：最流行的异步HTTP客户端
    • httpx：支持同步/异步，API更友好
    • asyncio：Python内置异步框架

  2. 并发控制：

    • 使用 Semaphore 控制并发数
    • 使用 asyncio.gather 并发执行

  3. 错误处理：

    • 捕获 ClientError 和 TimeoutError
    • 实现重试机制和指数退避

  4. 性能优化：

    • 复用 ClientSession
    • 使用连接池
    • 合理设置超时和并发数

这些案例涵盖了从基础到高级的异步HTTP请求用法,可以根据实际需求选择合适的实现方式。

上一篇Python多线程案例如何创建线程

下一篇Python协程并发案例如何多协程执行

相关文章

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