欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python中異步HTTP客戶端/服務(wù)器框架aiohttp的使用全面指南

 更新時(shí)間：2025年06月03日 10:11:50   作者：老胖閑聊  
aiohttp 是一個(gè)基于 Python asyncio 的異步 HTTP 客戶端/服務(wù)器框架,專為高性能網(wǎng)絡(luò)編程設(shè)計(jì),本文將為大家詳細(xì)介紹一下它的具體使用,需要的可以了解下

什么是 aiohttp

aiohttp 是一個(gè)基于 Python asyncio 的異步 HTTP 客戶端/服務(wù)器框架，專為高性能網(wǎng)絡(luò)編程設(shè)計(jì)。它提供了：

  • 異步 HTTP 客戶端（類似異步版 requests）
  • 異步 HTTP 服務(wù)器（類似異步版 Flask/Django）
  • 完整的 WebSocket 支持
  • 高效的連接池管理

核心優(yōu)勢(shì)

特性描述
異步非阻塞單線程處理數(shù)千并發(fā)連接
高性能遠(yuǎn)超同步框架（如 requests）的吞吐量
輕量級(jí)簡(jiǎn)潔的API，無復(fù)雜依賴
全面協(xié)議支持HTTP/1.1, HTTP/2(客戶端), WebSocket
生態(tài)完善良好文檔和活躍社區(qū)

基礎(chǔ)用法 - HTTP客戶端

安裝

pip install aiohttp

基本GET請(qǐng)求

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            print("狀態(tài)碼:", response.status)
            print("響應(yīng)內(nèi)容:", await response.text())

asyncio.run(main())

POST請(qǐng)求示例

async def post_example():
    async with aiohttp.ClientSession() as session:
        # 表單數(shù)據(jù)
        async with session.post(
            'https://httpbin.org/post', 
            data={'key': 'value'}
        ) as response:
            print(await response.json())
        
        # JSON數(shù)據(jù)
        async with session.post(
            'https://api.example.com/users',
            json={'name': 'Alice', 'age': 30}
        ) as response:
            print(await response.json())

高級(jí)用法

并發(fā)請(qǐng)求

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def concurrent_requests():
    urls = [
        'https://api.example.com/item/1',
        'https://api.example.com/item/2',
        'https://api.example.com/item/3'
    ]
    
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    for url, content in zip(urls, results):
        print(f"{url}: {content[:50]}...")

asyncio.run(concurrent_requests())

超時(shí)控制

async def timeout_example():
    timeout = aiohttp.ClientTimeout(total=5)  # 5秒總超時(shí)
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get('https://slow-api.example.com') as response:
                return await response.text()
        except asyncio.TimeoutError:
            print("請(qǐng)求超時(shí)!")

流式處理大響應(yīng)

async def stream_response():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://large-file.example.com') as response:
            with open('large_file.txt', 'wb') as f:
                async for chunk in response.content.iter_chunked(1024):
                    f.write(chunk)
                    print(f"已接收 {len(chunk)} 字節(jié)")

服務(wù)器端開發(fā)

基本HTTP服務(wù)器

from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "World")
    return web.Response(text=f"Hello, {name}!")

app = web.Application()
app.add_routes([
    web.get('/', handle),
    web.get('/{name}', handle)
])

if __name__ == '__main__':
    web.run_app(app, port=8080)

REST API示例

async def get_users(request):
    users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
    return web.json_response(users)

async def create_user(request):
    data = await request.json()
    # 實(shí)際應(yīng)用中這里會(huì)保存到數(shù)據(jù)庫(kù)
    return web.json_response({'id': 3, **data}, status=201)

app = web.Application()
app.add_routes([
    web.get('/api/users', get_users),
    web.post('/api/users', create_user)
])

WebSocket服務(wù)器

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            if msg.data == 'close':
                await ws.close()
            else:
                await ws.send_str(f"ECHO: {msg.data}")
        elif msg.type == aiohttp.WSMsgType.ERROR:
            print('WebSocket連接異常關(guān)閉')
    
    return ws

???????app.add_routes([web.get('/ws', websocket_handler)])

進(jìn)階擴(kuò)展

中間件示例

async def auth_middleware(app, handler):
    async def middleware(request):
        # 驗(yàn)證API密鑰
        if request.headers.get('X-API-Key') != 'SECRET_KEY':
            return web.json_response({'error': 'Unauthorized'}, status=401)
        return await handler(request)
    return middleware

app = web.Application(middlewares=[auth_middleware])

HTTP/2客戶端支持

async def http2_request():
    conn = aiohttp.TCPConnector(force_close=True, enable_cleanup_closed=True)
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.get(
            'https://http2.akamai.com/',
            headers={'accept': 'text/html'}
        ) as response:
            print("HTTP版本:", response.version)
            print("內(nèi)容:", await response.text()[:200])

性能優(yōu)化配置

# 自定義連接器配置
connector = aiohttp.TCPConnector(
    limit=100,  # 最大并發(fā)連接數(shù)
    limit_per_host=20,  # 單主機(jī)最大連接數(shù)
    ssl=False,  # 禁用SSL驗(yàn)證（僅用于測(cè)試）
    force_close=True  # 避免連接延遲關(guān)閉
)

# 自定義會(huì)話配置
session = aiohttp.ClientSession(
    connector=connector,
    timeout=aiohttp.ClientTimeout(total=30),
    headers={'User-Agent': 'MyApp/1.0'},
    cookie_jar=aiohttp.CookieJar(unsafe=True)
)

最佳實(shí)踐

重用ClientSession：避免為每個(gè)請(qǐng)求創(chuàng)建新會(huì)話

使用連接池：合理配置TCPConnector參數(shù)

超時(shí)設(shè)置：總是配置合理的超時(shí)時(shí)間

資源清理：使用async with確保資源釋放

錯(cuò)誤處理：捕獲并處理常見網(wǎng)絡(luò)異常

try:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.json()
except aiohttp.ClientError as e:
    print(f"請(qǐng)求錯(cuò)誤: {e}")

完整示例

import aiohttp
import asyncio
from aiohttp import web

# 客戶端示例
async def fetch_data():
    async with aiohttp.ClientSession() as session:
        # 并發(fā)請(qǐng)求多個(gè)API
        urls = [
            'https://jsonplaceholder.typicode.com/posts/1',
            'https://jsonplaceholder.typicode.com/comments/1',
            'https://jsonplaceholder.typicode.com/albums/1'
        ]
        
        tasks = []
        for url in urls:
            tasks.append(session.get(url))
        
        responses = await asyncio.gather(*tasks)
        
        results = []
        for response in responses:
            results.append(await response.json())
        
        return results

# 服務(wù)器示例
async def handle_index(request):
    return web.Response(text="Welcome to aiohttp server!")

async def handle_api(request):
    data = await fetch_data()
    return web.json_response(data)

# 創(chuàng)建應(yīng)用
app = web.Application()
app.add_routes([
    web.get('/', handle_index),
    web.get('/api', handle_api)
])

# 啟動(dòng)服務(wù)器
async def start_server():
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("Server running at http://localhost:8080")
    
    # 保持運(yùn)行
    while True:
        await asyncio.sleep(3600)  # 每小時(shí)喚醒一次

if __name__ == '__main__':
    asyncio.run(start_server())

總結(jié)

aiohttp 是 Python 異步生態(tài)中處理 HTTP 通信的首選工具，它提供了：

  • 高效客戶端：用于高性能爬蟲、API調(diào)用
  • 輕量級(jí)服務(wù)器：構(gòu)建高性能Web服務(wù)和API
  • WebSocket支持：實(shí)現(xiàn)實(shí)時(shí)雙向通信
  • 連接池管理：優(yōu)化資源利用率

通過合理利用 aiohttp 的異步特性，開發(fā)者可以輕松構(gòu)建出能夠處理數(shù)萬并發(fā)連接的高性能網(wǎng)絡(luò)應(yīng)用，同時(shí)保持代碼的簡(jiǎn)潔性和可維護(hù)性。

到此這篇關(guān)于Python中異步HTTP客戶端/服務(wù)器框架aiohttp的使用全面指南的文章就介紹到這了,更多相關(guān)Python異步框架aiohttp內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

最新評(píng)論