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

11-2. 全局變量陷阱（Global Variable Pitfalls）

📌 為什麼全局變量危險？

簡單說： 多個線程共享同一個全局變量，會互相干擾

核心問題：

🔴 數據混淆：線程 A 的數據被線程 B 覆蓋
🔴 競態條件：並發修改導致數據不一致
🔴 難以調試：問題隨機出現，難以復現

🔴 陷阱 1：模塊級變量

問題演示

# views.py

# ❌ 危險的全局變量
current_user = None

def login_view(request):
    global current_user
    current_user = request.user  # 設置全局變量
    return HttpResponse(f"Logged in as {current_user.username}")

def profile_view(request):
    global current_user
    # 期望獲取當前用戶
    username = current_user.username
    return HttpResponse(f"Profile of {username}")

問題場景

時間線：

T1    Thread 1: login_view()
      current_user = User("Alice")

T2    Thread 2: login_view()
      current_user = User("Bob")  ← 覆蓋了！

T3    Thread 1: profile_view()
      讀取 current_user
      → 得到 User("Bob") ❌（期望 Alice）

T4    Thread 2: profile_view()
      讀取 current_user
      → 得到 User("Bob") ✓

後果： Alice 看到了 Bob 的資料！嚴重的安全問題！

真實案例：購物車混亂

# cart.py

# ❌ 錯誤實現
shopping_cart = []  # 全局變量

def add_to_cart(request):
    product_id = request.POST.get('product_id')

    shopping_cart.append({
        'user': request.user.id,
        'product': product_id
    })

    return HttpResponse("Added to cart")

def view_cart(request):
    # 返回購物車內容
    user_items = [
        item for item in shopping_cart
        if item['user'] == request.user.id
    ]
    return JsonResponse({'items': user_items})

問題：

時間    用戶 A                  用戶 B                  shopping_cart
T1      加入商品 1              -                       [{'user': A, 'product': 1}]
T2      -                       加入商品 2              [{'user': A, 'product': 1},
                                                        {'user': B, 'product': 2}]
T3      查看購物車
        → 看到商品 1 ✓

T4      -                       查看購物車
                                → 看到商品 2 ✓

看起來正常？但是...

T5      用戶 C 加入商品 3
T6      服務器重啟 💥
T7      shopping_cart = []  ← 所有數據丟失！

🔴 陷阱 2：類變量

問題演示

# models.py

class RequestCounter:
    count = 0  # ❌ 類變量，所有實例共享

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

    @classmethod
    def get_count(cls):
        return cls.count

# views.py
def index(request):
    RequestCounter.increment()
    return HttpResponse(f"Total requests: {RequestCounter.get_count()}")

競態條件

# Thread 1 和 Thread 2 同時執行 increment()

Thread 1                    Thread 2
讀取 count = 100
                            讀取 count = 100
count + 1 = 101
                            count + 1 = 101
寫入 count = 101
                            寫入 count = 101

結果：count = 101（應該是 102）❌

問題： count += 1 不是原子操作，包含三個步驟：

讀取 count
計算 count + 1
寫回 count

真實案例：訂單號生成錯誤

# utils.py

class OrderNumberGenerator:
    _last_number = 0  # ❌ 類變量

    @classmethod
    def generate(cls):
        cls._last_number += 1
        return f"ORD{cls._last_number:08d}"

# views.py
def create_order(request):
    order_number = OrderNumberGenerator.generate()

    Order.objects.create(
        order_number=order_number,
        user=request.user
    )

    return JsonResponse({'order_number': order_number})

問題： 可能生成重複的訂單號！

Thread 1                    Thread 2
讀取 _last_number = 1000
                            讀取 _last_number = 1000
_last_number + 1 = 1001
                            _last_number + 1 = 1001
返回 "ORD00001001"
                            返回 "ORD00001001" ← 重複！

兩個訂單有相同的訂單號 💥

🔴 陷阱 3：可變默認參數

問題演示

# ❌ 危險的可變默認參數
def add_to_list(item, items=[]):  # [] 是可變對象！
    items.append(item)
    return items

# 使用
result1 = add_to_list(1)  # [1]
result2 = add_to_list(2)  # [1, 2] ← 意外！期望 [2]
result3 = add_to_list(3)  # [1, 2, 3] ← 累積了！

原因： 默認參數在函數定義時創建，所有調用共享同一個列表對象

真實案例：表單錯誤累積

# forms.py

# ❌ 錯誤實現
class MyForm:
    def __init__(self, data=None, errors=[]):  # 危險！
        self.data = data
        self.errors = errors

    def add_error(self, message):
        self.errors.append(message)

# views.py
def view1(request):
    form = MyForm()
    form.add_error("Error in view1")
    return render(request, 'form.html', {'form': form})

def view2(request):
    form = MyForm()
    # 沒有添加錯誤，但是...
    return render(request, 'form.html', {'form': form})

問題： view2 中的表單會顯示 view1 的錯誤！

用戶 A 訪問 view1:
  form.errors = ["Error in view1"]

用戶 B 訪問 view2:
  form.errors = ["Error in view1"] ← 意外！

🔴 陷阱 4：共享緩存字典

問題演示

# cache.py

# ❌ 錯誤的緩存實現
_cache = {}  # 全局字典

def get_user_data(user_id):
    if user_id in _cache:
        return _cache[user_id]

    user = User.objects.get(id=user_id)
    data = {
        'username': user.username,
        'email': user.email
    }

    _cache[user_id] = data  # 危險！
    return data

def clear_user_cache(user_id):
    if user_id in _cache:
        del _cache[user_id]

問題：

線程安全問題

# Thread 1 和 Thread 2 同時檢查
Thread 1: if user_id in _cache:  # False
Thread 2: if user_id in _cache:  # False

Thread 1: _cache[user_id] = data1
Thread 2: _cache[user_id] = data2  # 覆蓋！

內存洩漏

# _cache 永遠增長，從不清理
# 最終耗盡內存 💥

數據不一致

# 數據庫更新了，但緩存沒有
User.objects.filter(id=123).update(email='new@example.com')
# _cache[123] 仍然是舊的 email

✅ 正確的替代方案

方案 1：使用請求對象

# ✅ 正確：使用 request 對象傳遞數據

# middleware.py
class UserMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 將用戶數據附加到 request
        if request.user.is_authenticated:
            request.user_data = {
                'name': request.user.username,
                'email': request.user.email
            }

        response = self.get_response(request)
        return response

# views.py
def profile_view(request):
    # 從 request 獲取數據
    user_data = getattr(request, 'user_data', None)
    return HttpResponse(f"Email: {user_data['email']}")

方案 2：使用 threading.local

# ✅ 正確：使用線程本地存儲

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)

# middleware.py
class ThreadLocalUserMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        set_current_user(request.user)

        response = self.get_response(request)

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

        return response

# views.py
def profile_view(request):
    user = get_current_user()
    return HttpResponse(f"User: {user.username}")

方案 3：使用 Django 緩存

# ✅ 正確：使用 Django 緩存框架

from django.core.cache import cache

def get_user_data(user_id):
    cache_key = f'user_data:{user_id}'

    # Django 的 cache 是線程安全的
    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

def clear_user_cache(user_id):
    cache_key = f'user_data:{user_id}'
    cache.delete(cache_key)

方案 4：使用數據庫

# ✅ 正確：使用數據庫存儲狀態

# models.py
class ShoppingCart(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

class CartItem(models.Model):
    cart = models.ForeignKey(ShoppingCart, on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    quantity = models.IntegerField(default=1)

# views.py
def add_to_cart(request):
    product_id = request.POST.get('product_id')

    # 獲取或創建購物車
    cart, created = ShoppingCart.objects.get_or_create(user=request.user)

    # 添加商品
    CartItem.objects.create(
        cart=cart,
        product_id=product_id,
        quantity=1
    )

    return HttpResponse("Added to cart")

def view_cart(request):
    cart = ShoppingCart.objects.filter(user=request.user).first()
    if not cart:
        return JsonResponse({'items': []})

    items = cart.cartitem_set.select_related('product').all()
    return JsonResponse({
        'items': [
            {
                'product': item.product.name,
                'quantity': item.quantity
            }
            for item in items
        ]
    })

方案 5：使用原子操作

# ✅ 正確：使用數據庫原子操作

# models.py
class PageView(models.Model):
    url = models.CharField(max_length=200, unique=True)
    count = models.IntegerField(default=0)

# 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")

def get_page_views(url):
    try:
        page_view = PageView.objects.get(url=url)
        return page_view.count
    except PageView.DoesNotExist:
        return 0

🎯 實戰案例

案例 1：安全的訂單號生成

# ✅ 使用數據庫自增 ID

# models.py
class Order(models.Model):
    # 自增 ID 是線程安全的
    id = models.AutoField(primary_key=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

    @property
    def order_number(self):
        # 基於 ID 生成訂單號
        return f"ORD{self.created_at.year}{self.id:08d}"

# views.py
def create_order(request):
    order = Order.objects.create(user=request.user)

    return JsonResponse({
        'order_number': order.order_number  # 唯一且安全
    })

案例 2：安全的請求計數

# ✅ 使用 Redis 原子操作

from django.core.cache import cache

def track_api_usage(request):
    user_id = request.user.id
    date_key = timezone.now().strftime('%Y-%m-%d')
    cache_key = f'api_usage:{user_id}:{date_key}'

    # incr 是原子操作
    count = cache.incr(cache_key, default=0)

    # 設置過期時間（2 天）
    cache.touch(cache_key, timeout=172800)

    return count

def check_rate_limit(request):
    count = track_api_usage(request)

    if count > 1000:  # 每日限制 1000 次
        return JsonResponse({
            'error': 'Rate limit exceeded'
        }, status=429)

    # 繼續處理請求
    return JsonResponse({'status': 'ok'})

案例 3：安全的用戶會話數據

# ✅ 使用 Django Session

def store_user_preference(request):
    preference = request.POST.get('preference')

    # Session 是線程安全的
    request.session['preference'] = preference
    request.session.modified = True

    return HttpResponse("Saved")

def get_user_preference(request):
    preference = request.session.get('preference', 'default')
    return HttpResponse(f"Preference: {preference}")

💡 檢查清單

🔍 如何發現全局變量問題？

代碼審查

# 搜索這些模式：
# - 模塊級變量賦值
# - 類變量修改
# - 可變默認參數
# - global 關鍵字

單元測試

from threading import Thread

def test_thread_safety():
    """測試線程安全"""
    results = []

    def worker():
        # 執行可能有問題的代碼
        result = some_function()
        results.append(result)

    threads = [Thread(target=worker) for _ in range(10)]

    for t in threads:
        t.start()

    for t in threads:
        t.join()

    # 檢查結果是否一致
    assert len(set(results)) == 10, "數據混淆！"

壓力測試

# 使用 locust 或 ab 測試
ab -n 10000 -c 100 http://localhost:8000/api/endpoint/

💡 最佳實踐

1. 避免全局可變狀態

# ❌ 不好
cache = {}
current_user = None

# ✅ 好
# 使用 Django 提供的工具
from django.core.cache import cache
from django.contrib.auth import get_user

2. 使用不可變對象

# ✅ 不可變對象是線程安全的
ALLOWED_HOSTS = ['example.com', 'www.example.com']  # tuple
DEBUG = False  # bool
MAX_UPLOAD_SIZE = 5 * 1024 * 1024  # int

3. 明確文檔化

# ✅ 如果必須使用全局變量，明確文檔化

# 線程安全：使用 threading.Lock 保護
_counter_lock = threading.Lock()
_counter = 0

def increment_counter():
    """線程安全的計數器增加

    Warning: 使用全局變量，但已加鎖保護
    """
    global _counter
    with _counter_lock:
        _counter += 1

4. 定期審查

# 創建自定義 linter 規則
# .pylintrc
[MESSAGES CONTROL]
enable=global-statement,global-variable-not-assigned

💡 面試要點

Q1: 為什麼全局變量在 Django 中危險？

答：

Django 使用多線程處理請求
全局變量被所有線程共享
並發訪問導致數據混淆和競態條件
難以調試，問題隨機出現

Q2: 如何替代全局變量？

答：

Request 對象：傳遞請求相關數據
threading.local：線程本地存儲
Django Cache：緩存系統
數據庫：持久化狀態
Session：用戶會話數據

Q3: 類變量和實例變量的區別？

答：

類變量：所有實例共享，可能線程不安全
實例變量：每個實例獨立，通常線程安全（如果實例不共享）

Q4: 如何測試線程安全？

答：

多線程單元測試
並發壓力測試
代碼審查（搜索 global、類變量等）
使用靜態分析工具

🔗 下一篇

在下一篇文章中，我們將深入學習 競態條件處理，了解如何識別和解決各種競態條件問題。

閱讀時間：10 分鐘

Django 面試準備 11-2：全局變量陷阱

11-2. 全局變量陷阱（Global Variable Pitfalls）

📌 為什麼全局變量危險？

🔴 陷阱 1：模塊級變量

問題演示

問題場景

真實案例：購物車混亂

🔴 陷阱 2：類變量

問題演示

競態條件

真實案例：訂單號生成錯誤

🔴 陷阱 3：可變默認參數

問題演示

真實案例：表單錯誤累積

🔴 陷阱 4：共享緩存字典

問題演示

✅ 正確的替代方案

方案 1：使用請求對象

方案 2：使用 threading.local

方案 3：使用 Django 緩存

方案 4：使用數據庫

方案 5：使用原子操作

🎯 實戰案例

案例 1：安全的訂單號生成

案例 2：安全的請求計數

案例 3：安全的用戶會話數據

💡 檢查清單

🔍 如何發現全局變量問題？

💡 最佳實踐

1. 避免全局可變狀態

2. 使用不可變對象

3. 明確文檔化

4. 定期審查

💡 面試要點

Q1: 為什麼全局變量在 Django 中危險？

Q2: 如何替代全局變量？

Q3: 類變量和實例變量的區別？

Q4: 如何測試線程安全？

🔗 下一篇