2025-01-18 約 1300 字預計閱讀 6 分鐘

11-1. Django 線程安全機制（Thread Safety Mechanism）

📌 什麼是線程安全？

簡單說： 多個線程同時訪問同一資源時，不會出現數據錯誤或不一致

定義： 線程安全是指在多線程環境下，程序能夠正確地處理多個線程同時訪問共享資源，而不會導致數據競爭或不一致的狀態。

🔍 Django 的多線程環境

WSGI 服務器的線程模式

用戶請求 → WSGI 服務器 → Django 應用

WSGI 服務器（如 Gunicorn）：
┌────────────────────────────────┐
│ Worker Process 1               │
│  ├─ Thread 1 → Request A       │
│  ├─ Thread 2 → Request B       │
│  └─ Thread 3 → Request C       │
├────────────────────────────────┤
│ Worker Process 2               │
│  ├─ Thread 1 → Request D       │
│  ├─ Thread 2 → Request E       │
│  └─ Thread 3 → Request F       │
└────────────────────────────────┘

Django 請求處理流程

# 每個請求在獨立線程中處理

def handle_request(request):
    # 線程 1
    Thread-1: 處理 Request A
      ├─ 創建 HttpRequest 對象
      ├─ 執行中間件
      ├─ 調用 View 函數
      ├─ 渲染模板
      └─ 返回 HttpResponse

    # 線程 2（同時進行）
    Thread-2: 處理 Request B
      ├─ 創建 HttpRequest 對象
      ├─ 執行中間件
      ├─ 調用 View 函數
      ├─ 渲染模板
      └─ 返回 HttpResponse

關鍵： 每個請求都在獨立的線程中處理，互不干擾

✅ Django 的線程安全保證

1. 請求對象隔離

# views.py
def my_view(request):
    # request 對象是線程安全的
    # 每個請求都有獨立的 request 對象

    user = request.user  # 線程 1 的用戶
    # 不會與線程 2 的 request.user 混淆

    return HttpResponse("OK")

原理： Django 為每個請求創建新的 HttpRequest 對象

# Django 內部（簡化）
class WSGIHandler:
    def __call__(self, environ, start_response):
        # 每次調用都創建新對象
        request = self.request_class(environ)  # 新的 HttpRequest
        response = self.get_response(request)
        return response

2. 數據庫連接管理

# Django 自動管理數據庫連接

# Thread 1
def view1(request):
    users = User.objects.all()  # 使用 Thread 1 的連接
    return render(request, 'users.html', {'users': users})

# Thread 2（同時進行）
def view2(request):
    posts = Post.objects.all()  # 使用 Thread 2 的連接
    return render(request, 'posts.html', {'posts': posts})

連接池管理：

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'CONN_MAX_AGE': 600,  # 連接持續時間（秒）
        # Django 自動為每個線程維護獨立的連接
    }
}

原理： Django 使用 threading.local() 為每個線程存儲獨立的數據庫連接

# Django 內部（簡化）
class DatabaseWrapper:
    def __init__(self):
        self._connections = threading.local()  # 線程本地存儲

    def get_connection(self):
        # 每個線程獲取自己的連接
        if not hasattr(self._connections, 'connection'):
            self._connections.connection = self.create_connection()
        return self._connections.connection

3. 模板渲染

# 模板渲染是線程安全的

# Thread 1
def view1(request):
    return render(request, 'template.html', {'user': 'Alice'})

# Thread 2
def view2(request):
    return render(request, 'template.html', {'user': 'Bob'})

# 不會混淆！每個線程都有獨立的上下文

4. Session 管理

# Session 也是線程安全的

def view(request):
    # 每個請求都有獨立的 session
    request.session['user_id'] = request.user.id
    # 不會影響其他線程的 session

    return HttpResponse("OK")

原理： Session 數據存儲在數據庫/緩存中，通過 session_key 隔離

⚠️ 線程不安全的場景

場景 1：全局變量

# ❌ 線程不安全
current_user = None  # 全局變量

def login(request):
    global current_user
    current_user = request.user  # 危險！

    # 問題：
    # Thread 1: current_user = User A
    # Thread 2: current_user = User B  ← 覆蓋了 Thread 1
    # Thread 1 讀取: current_user → 得到 User B ❌

    return HttpResponse(f"Logged in as {current_user}")

後果：

時間    Thread 1                Thread 2
T1      current_user = Alice
T2                              current_user = Bob
T3      讀取 current_user
        → 得到 Bob ❌（應該是 Alice）

場景 2：類變量

# ❌ 線程不安全
class Counter:
    count = 0  # 類變量，所有實例共享

    @classmethod
    def increment(cls):
        cls.count += 1  # 非原子操作！

def view(request):
    Counter.increment()
    return HttpResponse(f"Count: {Counter.count}")

競態條件：

Thread 1                    Thread 2
讀取 count = 0
                            讀取 count = 0
count + 1 = 1
                            count + 1 = 1
寫入 count = 1
                            寫入 count = 1

結果：count = 1（應該是 2）❌

場景 3：共享可變對象

# ❌ 線程不安全
cache = {}  # 全局字典

def view(request):
    user_id = request.user.id

    if user_id not in cache:
        cache[user_id] = []  # 可能被多個線程同時修改

    cache[user_id].append(request.path)  # 危險！

    return HttpResponse("OK")

✅ 如何保證線程安全？

方法 1：使用線程本地存儲

import threading

# ✅ 線程安全
_thread_locals = threading.local()

def set_current_user(user):
    _thread_locals.user = user

def get_current_user():
    return getattr(_thread_locals, 'user', None)

# 使用
def view(request):
    set_current_user(request.user)
    # 每個線程都有獨立的 user
    user = get_current_user()
    return HttpResponse(f"User: {user}")

原理： threading.local() 為每個線程提供獨立的命名空間

方法 2：使用鎖

import threading

# ✅ 線程安全
class Counter:
    def __init__(self):
        self._count = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:  # 獲取鎖
            self._count += 1

    def get_count(self):
        with self._lock:
            return self._count

# 使用
counter = Counter()

def view(request):
    counter.increment()
    return HttpResponse(f"Count: {counter.get_count()}")

流程：

Thread 1                    Thread 2
獲取鎖 ✓
count += 1
                            嘗試獲取鎖（等待）
釋放鎖
                            獲取鎖 ✓
                            count += 1
                            釋放鎖

結果：count = 2 ✓

方法 3：使用原子操作

from django.core.cache import cache

# ✅ 使用 Redis 的原子操作
def view(request):
    # incr 是原子操作
    count = cache.incr('page_views', default=0)
    return HttpResponse(f"Views: {count}")

方法 4：避免共享狀態

# ✅ 最佳實踐：避免全局狀態

def view(request):
    # 使用局部變量
    user_data = {
        'id': request.user.id,
        'name': request.user.username
    }

    # 或使用 request 對象
    request.custom_data = {'key': 'value'}

    return HttpResponse("OK")

🎯 Django 內建的線程安全工具

1. threading.local 的使用

# Django 內部大量使用 threading.local

# 數據庫連接
from django.db import connection
# connection 對象使用 threading.local

# 當前語言設置
from django.utils.translation import get_language
# 每個線程可以有不同的語言設置

2. 請求中間件

# middleware.py
import threading

_thread_locals = threading.local()

class ThreadLocalMiddleware:
    """將 request 存儲到線程本地"""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _thread_locals.request = request
        response = self.get_response(request)
        # 清理
        if hasattr(_thread_locals, 'request'):
            del _thread_locals.request
        return response

def get_current_request():
    """在任何地方獲取當前請求"""
    return getattr(_thread_locals, 'request', None)

# 使用
def some_utility_function():
    request = get_current_request()
    if request:
        user = request.user
        # ...

3. Django Signals

# Signals 是線程安全的

from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=User)
def user_saved(sender, instance, created, **kwargs):
    # 這個函數在觸發信號的同一線程中執行
    # 不會有線程安全問題
    print(f"User {instance.username} saved")

📊 Gunicorn 工作模式對比

1. Sync Worker（同步）

gunicorn myapp.wsgi:application --workers 4

每個 worker 一次處理一個請求
┌─────────────┐
│ Worker 1    │ → Request A
├─────────────┤
│ Worker 2    │ → Request B
├─────────────┤
│ Worker 3    │ → Request C
├─────────────┤
│ Worker 4    │ → Request D
└─────────────┘

優點：簡單，無線程安全問題
缺點：並發低

2. Thread Worker（多線程）

gunicorn myapp.wsgi:application --workers 4 --threads 4

每個 worker 有多個線程
┌─────────────────────┐
│ Worker 1            │
│  ├─ Thread 1 → Req A│
│  ├─ Thread 2 → Req B│
│  ├─ Thread 3 → Req C│
│  └─ Thread 4 → Req D│
├─────────────────────┤
│ Worker 2            │
│  ├─ Thread 1 → Req E│
│  ├─ Thread 2 → Req F│
│  ├─ Thread 3 → Req G│
│  └─ Thread 4 → Req H│
└─────────────────────┘

優點：並發高，內存共享
缺點：需要注意線程安全

3. Gevent Worker（協程）

gunicorn myapp.wsgi:application --workers 4 --worker-class gevent

使用協程，不是真正的多線程
無線程安全問題（單線程事件循環）

🎯 實戰案例

案例 1：安全的計數器

# models.py
from django.db import models

class PageView(models.Model):
    url = models.CharField(max_length=200)
    count = models.IntegerField(default=0)

    class Meta:
        unique_together = ['url']

# views.py
from django.db.models import F

def track_page_view(request):
    url = request.path

    # ✅ 使用 F() 表達式，原子操作
    PageView.objects.update_or_create(
        url=url,
        defaults={'count': F('count') + 1}
    )

    return HttpResponse("OK")

案例 2：安全的用戶在線狀態

# middleware.py
import threading
from django.utils import timezone

_thread_locals = threading.local()

class CurrentUserMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _thread_locals.user = request.user

        # 更新最後活動時間（使用數據庫）
        if request.user.is_authenticated:
            User.objects.filter(id=request.user.id).update(
                last_active=timezone.now()
            )

        response = self.get_response(request)

        # 清理
        if hasattr(_thread_locals, 'user'):
            del _thread_locals.user

        return response

def get_current_user():
    return getattr(_thread_locals, 'user', None)

案例 3：安全的緩存使用

from django.core.cache import cache
import hashlib

def get_user_data(user_id):
    """線程安全的緩存獲取"""
    cache_key = f'user_data:{user_id}'

    # cache.get() 和 cache.set() 都是線程安全的
    data = cache.get(cache_key)

    if data is None:
        # 查詢數據庫
        user = User.objects.get(id=user_id)
        data = {
            'username': user.username,
            'email': user.email,
        }

        # 緩存 5 分鐘
        cache.set(cache_key, data, timeout=300)

    return data

💡 最佳實踐

1. 避免全局可變狀態

# ❌ 不好
cache_data = {}  # 全局字典

# ✅ 好：使用 Django 緩存
from django.core.cache import cache
cache.set('key', 'value')

2. 使用請求對象傳遞數據

# ❌ 不好
current_user = None

def middleware(request):
    global current_user
    current_user = request.user

# ✅ 好：使用 request 對象
def middleware(request):
    request.current_user = request.user

3. 使用數據庫事務

from django.db import transaction

# ✅ 使用事務保證原子性
@transaction.atomic
def transfer_money(from_user, to_user, amount):
    # 這些操作是原子的
    from_user.balance -= amount
    from_user.save()

    to_user.balance += amount
    to_user.save()

4. 謹慎使用鎖

import threading

lock = threading.Lock()

def critical_section():
    with lock:
        # 臨界區代碼
        # 盡量保持簡短
        pass

# ⚠️ 避免死鎖
# ❌ 不好
def func1():
    with lock1:
        with lock2:
            pass

def func2():
    with lock2:  # 不同的順序！
        with lock1:
            pass  # 可能死鎖

💡 面試要點

Q1: Django 是線程安全的嗎？

答：

核心是線程安全的：請求處理、ORM、模板渲染
但不是所有場景都安全：全局變量、類變量、共享可變對象
關鍵：避免在視圖中使用全局可變狀態

Q2: Django 如何保證請求隔離？

答：

每個請求創建新的 HttpRequest 對象
使用 threading.local() 隔離數據庫連接
Session 通過 session_key 隔離
中間件在每個請求獨立執行

Q3: 什麼情況下需要使用鎖？

答：

修改全局共享狀態
非原子的讀-改-寫操作
多個線程訪問同一文件
計數器等需要精確控制的場景

Q4: Gunicorn 的 workers 和 threads 如何選擇？

答：

CPU 密集：workers = CPU 核心數，threads = 1
I/O 密集：workers = 2-4，threads = 4-8
混合型：workers = CPU 核心數，threads = 2-4

🔗 下一篇

在下一篇文章中，我們將深入學習 全局變量陷阱，了解在 Django 中使用全局變量的各種問題和解決方案。

閱讀時間：8 分鐘

Django 面試準備 11-1：Django 線程安全機制