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 2

2. 異步視圖的定義與使用

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 何時使用異步視圖

推薦使用異步視圖

  1. I/O 密集型:大量數據庫查詢、外部 API 調用
  2. 高並發需求:需要處理大量並發請求
  3. 長時間等待:需要等待外部服務響應
  4. 實時功能:WebSocket、SSE

不推薦使用異步視圖

  1. CPU 密集型:圖像處理、數據分析(會阻塞事件循環)
  2. 依賴同步庫:無法異步化的第三方庫
  3. 簡單 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_asyncasync_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 密集型、簡單 CRUDI/O 密集型、高並發
數據庫查詢Model.objects.get()await Model.objects.aget()

Q2: 什麼時候應該使用異步視圖?

答案

應該使用異步視圖

  1. 需要並行執行多個 I/O 操作(數據庫、API 調用)
  2. 高並發場景(> 1000 QPS)
  3. 需要 WebSocket 或長連接
  4. 微服務架構(多個服務調用)

不應該使用異步視圖

  1. 簡單 CRUD(單個查詢)
  2. CPU 密集型任務(會阻塞事件循環)
  3. 依賴同步庫(無法異步化)

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 核心要點

  1. 異步視圖

    • 使用 async def 定義
    • 需要 ASGI 服務器(Uvicorn、Daphne)
    • Django 4.1+ 完整支持
  2. 性能優勢

    • I/O 密集型:提升 50-80%
    • 高並發:單個 Worker 處理數千請求
    • 並行執行:使用 asyncio.gather()
  3. 最佳實踐

    • ✅ I/O 密集型使用異步
    • ❌ CPU 密集型使用同步
    • ⚠️ 避免阻塞事件循環
  4. 常見陷阱

    • 在異步視圖中使用同步 ORM → 使用 aget(), asave()
    • CPU 密集型操作 → 使用 loop.run_in_executor()
    • 忘記 await → 導致返回協程對象

9.2 快速決策表

場景推薦理由
多個並行 I/O 操作異步 ✅性能提升明顯
單個數據庫查詢同步 ✅更簡單,性能相當
CPU 密集型計算同步 ✅異步會阻塞事件循環
微服務調用異步 ✅並行調用多個服務
WebSocket異步 ✅長連接必須用異步
簡單 CRUD同步 ✅沒必要增加複雜度

參考資料

  1. 官方文檔

  2. 第三方庫

  3. 深入閱讀


下一篇預告:12-3. 異步 ORM 查詢 - 深入探討 Django 異步 ORM 的實現原理與高級用法

0%