Django 面試精華:Django Async Views
從同步到異步:掌握 Django 異步視圖的最佳實踐
前言
Django 3.0 開始支持異步視圖,Django 3.1 正式穩定。這為 I/O 密集型應用帶來了巨大的性能提升。
傳統同步視圖的問題:
def user_dashboard(request):
# 串行執行,總耗時相加
user = User.objects.get(id=request.user.id) # 50ms
orders = Order.objects.filter(user=user).all() # 100ms
notifications = Notification.objects.filter(user=user).all() # 80ms
# 總耗時:230ms
return render(request, 'dashboard.html', {
'user': user,
'orders': orders,
'notifications': notifications,
})異步視圖的優勢:
async def user_dashboard(request):
# 並行執行,總耗時取最慢的那個
user, orders, notifications = await asyncio.gather(
User.objects.aget(id=request.user.id), # 50ms
Order.objects.filter(user=user).all(), # 100ms
Notification.objects.filter(user=user).all(), # 80ms
)
# 總耗時:100ms(取最慢的那個)
return render(request, 'dashboard.html', {
'user': user,
'orders': orders,
'notifications': notifications,
})性能提升:響應時間減少 56%(230ms → 100ms)!
這篇文章將深入探討 Django 異步視圖的實現、最佳實踐和常見陷阱。
1. Django 異步視圖基礎
1.1 什麼是異步視圖
異步視圖是使用 async def 定義的視圖函數,支持 await 關鍵字來執行異步操作。
語法對比:
# 同步視圖(傳統)
def sync_view(request):
user = User.objects.get(id=1)
return HttpResponse(f"Hello {user.name}")
# 異步視圖(Django 3.0+)
async def async_view(request):
user = await User.objects.aget(id=1) # 注意:aget 是異步方法
return HttpResponse(f"Hello {user.name}")關鍵差異:
| 特性 | 同步視圖 | 異步視圖 |
|---|---|---|
| 定義方式 | def view(request) | async def view(request) |
| 數據庫查詢 | Model.objects.get() | await Model.objects.aget() |
| 執行模型 | 阻塞線程 | 非阻塞事件循環 |
| 適用場景 | CPU 密集型、傳統應用 | I/O 密集型、高並發 |
1.2 Django 版本支持
| Django 版本 | 異步視圖支持 | 異步 ORM 支持 |
|---|---|---|
| < 3.0 | ❌ | ❌ |
| 3.0 | 🟡 實驗性 | ❌ |
| 3.1 | ✅ 穩定 | ❌ |
| 3.2 - 4.0 | ✅ | 🟡 部分支持 |
| 4.1+ | ✅ | ✅ 完整支持 |
推薦:使用 Django 4.1+ 獲得最佳異步體驗。
1.3 運行環境要求
異步視圖只能在 ASGI 服務器下運行:
# ❌ 錯誤:WSGI 不支持異步視圖
gunicorn myproject.wsgi:application
# ✅ 正確:ASGI 服務器
uvicorn myproject.asgi:application
# ✅ 推薦:Gunicorn + Uvicorn Workers
gunicorn myproject.asgi:application \
-k uvicorn.workers.UvicornWorker \
--workers 22. 異步視圖的定義與使用
2.1 基本語法
最簡單的異步視圖:
# views.py
from django.http import HttpResponse
async def hello_async(request):
return HttpResponse("Hello Async World!")配置路由:
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('hello/', views.hello_async), # 與同步視圖配置方式相同
]2.2 異步數據庫查詢
Django 4.1+ 提供了異步 ORM API:
常用異步方法:
| 同步方法 | 異步方法 | 說明 |
|---|---|---|
.get() | .aget() | 獲取單個對象 |
.create() | .acreate() | 創建對象 |
.save() | .asave() | 保存對象 |
.delete() | .adelete() | 刪除對象 |
.filter() | .filter() (返回 AsyncQuerySet) | 過濾查詢 |
.all() | .all() (返回 AsyncQuerySet) | 獲取所有對象 |
.count() | .acount() | 計數 |
.exists() | .aexists() | 檢查是否存在 |
示例:
from django.http import JsonResponse
from .models import User, Post
async def user_detail(request, user_id):
# 異步獲取用戶
user = await User.objects.aget(id=user_id)
# 異步查詢文章列表
posts = []
async for post in Post.objects.filter(user=user):
posts.append({
'title': post.title,
'created_at': post.created_at,
})
return JsonResponse({
'user': {
'id': user.id,
'name': user.name,
},
'posts': posts,
})2.3 並行執行多個查詢
使用 asyncio.gather() 並行執行多個異步操作:
import asyncio
from django.http import JsonResponse
from .models import User, Post, Comment
async def user_dashboard(request):
user_id = request.user.id
# 並行執行三個查詢
user, posts, comments = await asyncio.gather(
User.objects.aget(id=user_id),
Post.objects.filter(user_id=user_id).acount(),
Comment.objects.filter(user_id=user_id).acount(),
)
return JsonResponse({
'user': user.name,
'post_count': posts,
'comment_count': comments,
})性能對比:
# 同步版本(串行)
def sync_dashboard(request):
user = User.objects.get(id=request.user.id) # 50ms
post_count = Post.objects.filter(user=user).count() # 80ms
comment_count = Comment.objects.filter(user=user).count() # 70ms
# 總耗時:200ms
# 異步版本(並行)
async def async_dashboard(request):
user, post_count, comment_count = await asyncio.gather(
User.objects.aget(id=request.user.id), # 50ms
Post.objects.filter(user=user).acount(), # 80ms
Comment.objects.filter(user=user).acount(), # 70ms
)
# 總耗時:80ms(取最慢的那個)
# 性能提升:60%2.4 異步外部 API 調用
使用 aiohttp 進行異步 HTTP 請求:
import aiohttp
from django.http import JsonResponse
async def fetch_external_data(request):
async with aiohttp.ClientSession() as session:
# 並行調用多個外部 API
results = await asyncio.gather(
fetch_api(session, 'https://api1.example.com/data'),
fetch_api(session, 'https://api2.example.com/data'),
fetch_api(session, 'https://api3.example.com/data'),
)
return JsonResponse({'data': results})
async def fetch_api(session, url):
"""輔助函數:異步獲取 API 數據"""
async with session.get(url) as response:
return await response.json()與同步版本對比:
import requests
# 同步版本
def sync_fetch_external_data(request):
# 串行執行,每個請求 500ms
result1 = requests.get('https://api1.example.com/data').json() # 500ms
result2 = requests.get('https://api2.example.com/data').json() # 500ms
result3 = requests.get('https://api3.example.com/data').json() # 500ms
# 總耗時:1500ms
return JsonResponse({'data': [result1, result2, result3]})
# 異步版本
async def async_fetch_external_data(request):
async with aiohttp.ClientSession() as session:
results = await asyncio.gather(
fetch_api(session, 'https://api1.example.com/data'), # 500ms
fetch_api(session, 'https://api2.example.com/data'), # 500ms
fetch_api(session, 'https://api3.example.com/data'), # 500ms
)
# 總耗時:500ms(並行執行)
# 性能提升:66%
return JsonResponse({'data': results})3. 類視圖(Class-Based Views)的異步支持
3.1 異步類視圖
Django 3.1+ 支持異步類視圖:
from django.views import View
from django.http import JsonResponse
class AsyncUserView(View):
async def get(self, request, user_id):
"""異步 GET 方法"""
user = await User.objects.aget(id=user_id)
return JsonResponse({'user': user.name})
async def post(self, request, user_id):
"""異步 POST 方法"""
user = await User.objects.aget(id=user_id)
user.name = request.POST.get('name')
await user.asave()
return JsonResponse({'success': True})路由配置:
# urls.py
urlpatterns = [
path('users/<int:user_id>/', AsyncUserView.as_view()),
]3.2 異步通用視圖
Django 4.1+ 的通用視圖也支持異步:
from django.views.generic import ListView, DetailView
from .models import Post
# 異步列表視圖
class AsyncPostListView(ListView):
model = Post
template_name = 'posts/list.html'
async def get_queryset(self):
"""異步獲取查詢集"""
posts = []
async for post in Post.objects.filter(published=True):
posts.append(post)
return posts
# 異步詳情視圖
class AsyncPostDetailView(DetailView):
model = Post
template_name = 'posts/detail.html'
async def get_object(self, queryset=None):
"""異步獲取對象"""
post_id = self.kwargs.get('pk')
return await Post.objects.aget(id=post_id)4. 異步視圖的最佳實踐
4.1 何時使用異步視圖
✅ 推薦使用異步視圖:
- I/O 密集型:大量數據庫查詢、外部 API 調用
- 高並發需求:需要處理大量並發請求
- 長時間等待:需要等待外部服務響應
- 實時功能:WebSocket、SSE
❌ 不推薦使用異步視圖:
- CPU 密集型:圖像處理、數據分析(會阻塞事件循環)
- 依賴同步庫:無法異步化的第三方庫
- 簡單 CRUD:沒有並發操作的簡單視圖
決策流程:
視圖需要執行多個 I/O 操作?
├─ Yes → 這些操作可以並行執行嗎?
│ ├─ Yes → 使用異步視圖 ✅
│ └─ No → 使用同步視圖
│
└─ No → 是否有高並發需求?
├─ Yes → 考慮使用異步視圖
└─ No → 使用同步視圖 ✅4.2 避免阻塞事件循環
❌ 錯誤示例:在異步視圖中使用同步 API
async def bad_async_view(request):
# ❌ 阻塞事件循環(同步數據庫查詢)
user = User.objects.get(id=1) # 這會阻塞整個事件循環!
# ❌ 阻塞事件循環(同步 HTTP 請求)
response = requests.get('https://api.example.com/data') # 阻塞!
# ❌ 阻塞事件循環(CPU 密集型操作)
result = complex_calculation() # 阻塞!
return HttpResponse("Bad!")✅ 正確示例:使用異步 API
import aiohttp
import asyncio
async def good_async_view(request):
# ✅ 使用異步數據庫查詢
user = await User.objects.aget(id=1)
# ✅ 使用異步 HTTP 請求
async with aiohttp.ClientSession() as session:
async with session.get('https://api.example.com/data') as response:
data = await response.json()
# ✅ CPU 密集型操作放到線程池執行
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, complex_calculation)
return HttpResponse("Good!")4.3 混合同步和異步代碼
使用 sync_to_async 調用同步代碼:
from asgiref.sync import sync_to_async
# 同步函數
def sync_function():
return User.objects.all() # 同步查詢
# 在異步視圖中調用
async def async_view(request):
# 將同步函數轉換為異步
users = await sync_to_async(sync_function)()
return JsonResponse({'count': len(users)})
# 或者使用裝飾器
@sync_to_async
def get_users():
return list(User.objects.all())
async def async_view2(request):
users = await get_users()
return JsonResponse({'count': len(users)})使用 async_to_sync 在同步代碼中調用異步函數:
from asgiref.sync import async_to_sync
# 異步函數
async def async_fetch_data():
async with aiohttp.ClientSession() as session:
async with session.get('https://api.example.com/data') as response:
return await response.json()
# 在同步視圖中調用
def sync_view(request):
# 將異步函數轉換為同步
data = async_to_sync(async_fetch_data)()
return JsonResponse(data)⚠️ 注意:頻繁使用 sync_to_async 和 async_to_sync 會降低性能,應盡量保持代碼風格一致。
4.4 錯誤處理
異步視圖的錯誤處理與同步視圖類似:
from django.http import JsonResponse, HttpResponseNotFound
from django.core.exceptions import ObjectDoesNotExist
async def async_user_detail(request, user_id):
try:
# 嘗試獲取用戶
user = await User.objects.aget(id=user_id)
# 並行獲取相關數據
posts, comments = await asyncio.gather(
Post.objects.filter(user=user).acount(),
Comment.objects.filter(user=user).acount(),
return_exceptions=True, # 不讓一個錯誤影響其他操作
)
# 檢查是否有錯誤
if isinstance(posts, Exception):
posts = 0
if isinstance(comments, Exception):
comments = 0
return JsonResponse({
'user': user.name,
'posts': posts,
'comments': comments,
})
except ObjectDoesNotExist:
return HttpResponseNotFound("User not found")
except Exception as e:
return JsonResponse({'error': str(e)}, status=500)5. 性能對比與測試
5.1 基準測試
測試場景:獲取用戶資料,需要查詢 3 個不同的表。
同步視圖:
def sync_user_dashboard(request, user_id):
import time
start = time.time()
# 串行執行
user = User.objects.get(id=user_id) # 50ms
posts = Post.objects.filter(user=user).count() # 80ms
comments = Comment.objects.filter(user=user).count() # 70ms
elapsed = (time.time() - start) * 1000
print(f"同步視圖耗時: {elapsed:.2f}ms") # ~200ms
return JsonResponse({
'user': user.name,
'posts': posts,
'comments': comments,
})異步視圖:
async def async_user_dashboard(request, user_id):
import time
start = time.time()
# 並行執行
user, posts, comments = await asyncio.gather(
User.objects.aget(id=user_id), # 50ms
Post.objects.filter(user_id=user_id).acount(), # 80ms
Comment.objects.filter(user_id=user_id).acount(), # 70ms
)
elapsed = (time.time() - start) * 1000
print(f"異步視圖耗時: {elapsed:.2f}ms") # ~80ms
return JsonResponse({
'user': user.name,
'posts': posts,
'comments': comments,
})結果對比:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
指標 │ 同步視圖 │ 異步視圖 │ 提升
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
響應時間 │ 200ms │ 80ms │ 60% ↓
吞吐量 (QPS) │ 50 │ 125 │ 150% ↑
Worker 占用 │ 200ms │ ~1ms │ 99% ↓
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━5.2 壓力測試
使用 wrk 進行壓力測試:
# 測試同步視圖
wrk -t4 -c100 -d30s http://localhost:8000/sync/user/1/
# Running 30s test @ http://localhost:8000/sync/user/1/
# 4 threads and 100 connections
# Requests/sec: 520.43
# Transfer/sec: 95.23KB
# 測試異步視圖
wrk -t4 -c100 -d30s http://localhost:8000/async/user/1/
# Running 30s test @ http://localhost:8000/async/user/1/
# 4 threads and 100 connections
# Requests/sec: 1342.17
# Transfer/sec: 245.78KB結論:異步視圖的吞吐量提升了 158%!
6. 常見陷阱與解決方案
6.1 陷阱 1:在異步視圖中使用同步 ORM
❌ 錯誤:
async def bad_view(request):
# SynchronousOnlyOperation 錯誤!
users = User.objects.all() # 同步查詢
return JsonResponse({'count': len(users)})錯誤信息:
SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.✅ 解決方案 1:使用異步 ORM API
async def good_view(request):
users = []
async for user in User.objects.all():
users.append(user)
return JsonResponse({'count': len(users)})✅ 解決方案 2:使用 sync_to_async
from asgiref.sync import sync_to_async
async def good_view(request):
users = await sync_to_async(list)(User.objects.all())
return JsonResponse({'count': len(users)})6.2 陷阱 2:CPU 密集型操作阻塞事件循環
❌ 錯誤:
async def bad_view(request):
# CPU 密集型操作,阻塞事件循環 1 秒
result = complex_calculation() # 1 秒
# 在這 1 秒內,整個事件循環被阻塞,無法處理其他請求!
return JsonResponse({'result': result})✅ 解決方案:使用線程池執行 CPU 密集型操作
import asyncio
async def good_view(request):
loop = asyncio.get_event_loop()
# 在線程池中執行 CPU 密集型操作
result = await loop.run_in_executor(None, complex_calculation)
return JsonResponse({'result': result})6.3 陷阱 3:忘記 await
❌ 錯誤:
async def bad_view(request):
# 忘記 await,返回的是協程對象,不是結果!
user = User.objects.aget(id=1) # <coroutine object>
print(user.name) # AttributeError: 'coroutine' object has no attribute 'name'
return HttpResponse("Bad!")✅ 解決方案:記得加 await
async def good_view(request):
user = await User.objects.aget(id=1) # 正確
print(user.name) # OK
return HttpResponse("Good!")6.4 陷阱 4:過度使用異步
❌ 錯誤:簡單 CRUD 也用異步
# 不必要的異步視圖
async def bad_list_view(request):
posts = []
async for post in Post.objects.all():
posts.append(post)
return render(request, 'posts.html', {'posts': posts})
# 只有一個查詢,沒有並發操作,異步沒有優勢!✅ 解決方案:簡單場景用同步
# 同步視圖更簡單、性能相當
def good_list_view(request):
posts = Post.objects.all()
return render(request, 'posts.html', {'posts': posts})原則:
- ✅ 多個 I/O 操作可以並行 → 使用異步
- ❌ 單個 I/O 操作 → 使用同步(更簡單)
7. 實戰案例
7.1 案例 1:用戶儀表板(並行查詢)
from django.http import JsonResponse
from .models import User, Order, Notification
import asyncio
async def user_dashboard_api(request):
"""用戶儀表板 API - 並行查詢多個數據源"""
user_id = request.user.id
# 並行執行 5 個查詢
results = await asyncio.gather(
# 基本資料
User.objects.aget(id=user_id),
# 訂單統計
Order.objects.filter(user_id=user_id).acount(),
Order.objects.filter(user_id=user_id, status='pending').acount(),
# 通知統計
Notification.objects.filter(user_id=user_id, read=False).acount(),
# 最近訂單
get_recent_orders(user_id),
)
user, total_orders, pending_orders, unread_notifications, recent_orders = results
return JsonResponse({
'user': {
'id': user.id,
'name': user.name,
'email': user.email,
},
'stats': {
'total_orders': total_orders,
'pending_orders': pending_orders,
'unread_notifications': unread_notifications,
},
'recent_orders': recent_orders,
})
async def get_recent_orders(user_id, limit=5):
"""獲取最近訂單"""
orders = []
async for order in Order.objects.filter(user_id=user_id).order_by('-created_at')[:limit]:
orders.append({
'id': order.id,
'total': float(order.total),
'status': order.status,
'created_at': order.created_at.isoformat(),
})
return orders性能對比:
- 同步版本:5 個查詢串行執行,假設每個 50ms,總計 250ms
- 異步版本:5 個查詢並行執行,總計 ~50ms
- 性能提升:80%
7.2 案例 2:聚合多個微服務數據
import aiohttp
from django.http import JsonResponse
async def aggregated_data_api(request):
"""聚合多個微服務的數據"""
user_id = request.user.id
async with aiohttp.ClientSession() as session:
# 並行調用 4 個微服務
results = await asyncio.gather(
fetch_user_service(session, user_id),
fetch_order_service(session, user_id),
fetch_payment_service(session, user_id),
fetch_recommendation_service(session, user_id),
return_exceptions=True, # 容錯:單個服務失敗不影響其他
)
user_data, order_data, payment_data, recommendation_data = results
# 處理錯誤
if isinstance(user_data, Exception):
user_data = {'error': 'User service unavailable'}
if isinstance(order_data, Exception):
order_data = {'orders': []}
return JsonResponse({
'user': user_data,
'orders': order_data,
'payment': payment_data,
'recommendations': recommendation_data,
})
async def fetch_user_service(session, user_id):
"""調用用戶服務"""
async with session.get(f'http://user-service/api/users/{user_id}') as response:
return await response.json()
async def fetch_order_service(session, user_id):
"""調用訂單服務"""
async with session.get(f'http://order-service/api/orders?user_id={user_id}') as response:
return await response.json()
async def fetch_payment_service(session, user_id):
"""調用支付服務"""
async with session.get(f'http://payment-service/api/payments?user_id={user_id}') as response:
return await response.json()
async def fetch_recommendation_service(session, user_id):
"""調用推薦服務"""
async with session.get(f'http://recommendation-service/api/recommendations/{user_id}') as response:
return await response.json()性能對比:
- 同步版本:4 個微服務串行調用,每個 300ms,總計 1200ms
- 異步版本:4 個微服務並行調用,總計 ~300ms
- 性能提升:75%
7.3 案例 3:文件上傳與處理
import aiofiles
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
async def async_file_upload(request):
"""異步處理文件上傳"""
if request.method != 'POST':
return JsonResponse({'error': 'Method not allowed'}, status=405)
# 異步讀取上傳的文件
file = request.FILES.get('file')
if not file:
return JsonResponse({'error': 'No file provided'}, status=400)
# 異步寫入文件
file_path = f'/tmp/{file.name}'
async with aiofiles.open(file_path, 'wb') as f:
# 分塊寫入
for chunk in file.chunks():
await f.write(chunk)
# 並行執行多個處理任務
results = await asyncio.gather(
analyze_file(file_path), # 分析文件內容
generate_thumbnail(file_path), # 生成縮略圖
upload_to_s3(file_path), # 上傳到 S3
)
analysis, thumbnail, s3_url = results
return JsonResponse({
'success': True,
'file_path': file_path,
'analysis': analysis,
'thumbnail': thumbnail,
's3_url': s3_url,
})
async def analyze_file(file_path):
"""分析文件(模擬)"""
await asyncio.sleep(0.5) # 模擬耗時操作
return {'type': 'image', 'size': 1024}
async def generate_thumbnail(file_path):
"""生成縮略圖(模擬)"""
await asyncio.sleep(0.3)
return '/thumbnails/thumb.jpg'
async def upload_to_s3(file_path):
"""上傳到 S3(模擬)"""
await asyncio.sleep(0.4)
return 'https://s3.amazonaws.com/bucket/file.jpg'8. 面試常見問題
Q1: Django 異步視圖和同步視圖有什麼區別?
答案:
| 特性 | 同步視圖 | 異步視圖 |
|---|---|---|
| 定義方式 | def view(request) | async def view(request) |
| 執行模型 | 阻塞線程 | 非阻塞事件循環 |
| 並發處理 | 受限於 Worker 數量 | 單個 Worker 處理數千請求 |
| 適用場景 | CPU 密集型、簡單 CRUD | I/O 密集型、高並發 |
| 數據庫查詢 | Model.objects.get() | await Model.objects.aget() |
Q2: 什麼時候應該使用異步視圖?
答案:
✅ 應該使用異步視圖:
- 需要並行執行多個 I/O 操作(數據庫、API 調用)
- 高並發場景(> 1000 QPS)
- 需要 WebSocket 或長連接
- 微服務架構(多個服務調用)
❌ 不應該使用異步視圖:
- 簡單 CRUD(單個查詢)
- CPU 密集型任務(會阻塞事件循環)
- 依賴同步庫(無法異步化)
Q3: 異步視圖一定比同步視圖快嗎?
答案:
不一定,取決於具體場景:
異步更快(I/O 密集型):
# 多個 I/O 操作並行執行
async def fast_view(request):
data = await asyncio.gather(
db_query(), # 100ms
api_call(), # 200ms
)
# 總耗時:200ms
def slow_view(request):
data1 = db_query() # 100ms
data2 = api_call() # 200ms
# 總耗時:300ms同步可能更快(單個操作):
# 單個查詢,異步有額外開銷
async def async_view(request):
user = await User.objects.aget(id=1) # 50ms + 協程開銷
return JsonResponse({'user': user.name})
def sync_view(request):
user = User.objects.get(id=1) # 50ms
return JsonResponse({'user': user.name})Q4: 如何在異步視圖中使用同步代碼?
答案:
使用 sync_to_async:
from asgiref.sync import sync_to_async
# 方式 1:裝飾器
@sync_to_async
def sync_function():
return User.objects.all()
async def async_view(request):
users = await sync_function()
# 方式 2:直接調用
async def async_view2(request):
users = await sync_to_async(list)(User.objects.all())注意:過度使用會降低性能。
Q5: Django 異步 ORM 支持哪些操作?
答案(Django 4.1+):
✅ 支持的操作:
.aget(),.acreate(),.asave(),.adelete().acount(),.aexists(),.aget_or_create()async for遍歷查詢集.filter(),.exclude(),.order_by()等(返回 AsyncQuerySet)
❌ 不支持的操作:
- 聚合函數(
.aggregate())(Django 5.0+ 支持) - 事務(
transaction.atomic)(Django 4.2+ 部分支持) - 信號(Signals)
9. 總結
9.1 核心要點
異步視圖:
- 使用
async def定義 - 需要 ASGI 服務器(Uvicorn、Daphne)
- Django 4.1+ 完整支持
- 使用
性能優勢:
- I/O 密集型:提升 50-80%
- 高並發:單個 Worker 處理數千請求
- 並行執行:使用
asyncio.gather()
最佳實踐:
- ✅ I/O 密集型使用異步
- ❌ CPU 密集型使用同步
- ⚠️ 避免阻塞事件循環
常見陷阱:
- 在異步視圖中使用同步 ORM → 使用
aget(),asave()等 - CPU 密集型操作 → 使用
loop.run_in_executor() - 忘記
await→ 導致返回協程對象
- 在異步視圖中使用同步 ORM → 使用
9.2 快速決策表
| 場景 | 推薦 | 理由 |
|---|---|---|
| 多個並行 I/O 操作 | 異步 ✅ | 性能提升明顯 |
| 單個數據庫查詢 | 同步 ✅ | 更簡單,性能相當 |
| CPU 密集型計算 | 同步 ✅ | 異步會阻塞事件循環 |
| 微服務調用 | 異步 ✅ | 並行調用多個服務 |
| WebSocket | 異步 ✅ | 長連接必須用異步 |
| 簡單 CRUD | 同步 ✅ | 沒必要增加複雜度 |
參考資料
官方文檔:
第三方庫:
深入閱讀:
下一篇預告:12-3. 異步 ORM 查詢 - 深入探討 Django 異步 ORM 的實現原理與高級用法