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

wen python案例 2026-06-30 200

本文目录导读：

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

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

我来为你介绍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())

异步库选择：
- aiohttp：最流行的异步HTTP客户端
- httpx：支持同步/异步，API更友好
- asyncio：Python内置异步框架
并发控制：
- 使用 Semaphore 控制并发数
- 使用 asyncio.gather 并发执行
错误处理：
- 捕获 ClientError 和 TimeoutError
- 实现重试机制和指数退避
性能优化：
- 复用 ClientSession
- 使用连接池
- 合理设置超时和并发数

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

本文地址： https://www.bmjup.com/post/12814.html