Django 面試精華：threading.local 原理

# ❌ 全局變量會被所有線程共享
current_user = None

def process_request(request):
    global current_user
    current_user = request.user  # 線程 A 設置

    do_some_work()  # 在這期間，線程 B 可能修改 current_user！

    send_email(current_user)  # 可能發給錯誤的用戶！

# ✅ 每個線程有自己的 current_user
thread_local = threading.local()

def process_request(request):
    thread_local.current_user = request.user  # 線程隔離

    do_some_work()  # 其他線程不會影響

    send_email(thread_local.current_user)  # 安全！

import threading

class SimpleLocal:
    def __init__(self):
        # 使用字典存儲，key 是線程 ID
        self._storage = {}

    def _get_thread_id(self):
        return threading.current_thread().ident

    def __setattr__(self, name, value):
        if name == '_storage':
            # 初始化 _storage 本身
            object.__setattr__(self, name, value)
        else:
            # 存儲到當前線程的空間
            thread_id = self._get_thread_id()
            if thread_id not in self._storage:
                self._storage[thread_id] = {}
            self._storage[thread_id][name] = value

    def __getattr__(self, name):
        thread_id = self._get_thread_id()
        if thread_id in self._storage:
            return self._storage[thread_id].get(name)
        raise AttributeError(f"No such attribute: {name}")


# 使用示例
local_data = SimpleLocal()

def worker(name):
    # 每個線程設置自己的變量
    local_data.user = name
    local_data.count = 0

    for i in range(3):
        local_data.count += 1
        print(f"線程 {name}: count = {local_data.count}")

# 啟動多個線程
import concurrent.futures

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(worker, ['Alice', 'Bob', 'Charlie'])

# 輸出：
# 線程 Alice: count = 1
# 線程 Bob: count = 1
# 線程 Charlie: count = 1
# 線程 Alice: count = 2
# 線程 Bob: count = 2
# 線程 Charlie: count = 2
# 線程 Alice: count = 3
# 線程 Bob: count = 3
# 線程 Charlie: count = 3

# Lib/threading.py (簡化版本)

class local:
    __slots__ = '_local__impl', '_local__lock', '__dict__'

    def __new__(cls, *args, **kwargs):
        # 每個 local 對象都有自己的實現
        self = object.__new__(cls)
        impl = _localimpl()
        impl.localargs = (args, kwargs)
        impl.locallock = RLock()
        object.__setattr__(self, '_local__impl', impl)
        object.__setattr__(self, '_local__lock', impl.locallock)

        # 註冊當前線程
        impl.create_dict()
        return self

    def __getattribute__(self, name):
        # 獲取當前線程的字典
        impl = object.__getattribute__(self, '_local__impl')
        dct = impl.get_dict()
        return dct[name]

    def __setattr__(self, name, value):
        # 設置到當前線程的字典
        impl = object.__getattribute__(self, '_local__impl')
        dct = impl.get_dict()
        dct[name] = value


class _localimpl:
    def get_dict(self):
        # 獲取當前線程的字典
        thread_id = get_ident()
        return self.dicts.get(thread_id, {})

    def create_dict(self):
        # 為當前線程創建字典
        thread_id = get_ident()
        d = {}
        self.dicts[thread_id] = d
        return d

# django/db/backends/base/base.py (簡化)

class BaseDatabaseWrapper:
    def __init__(self, settings_dict, alias=DEFAULT_DB_ALIAS):
        # 使用 threading.local 存儲連接
        self._thread_local = threading.local()
        self.settings_dict = settings_dict
        self.alias = alias

    @property
    def connection(self):
        """獲取當前線程的資料庫連接"""
        # 檢查當前線程是否已有連接
        if not hasattr(self._thread_local, 'connection'):
            # 如果沒有，創建新連接
            self._thread_local.connection = self.get_new_connection(
                self.get_connection_params()
            )
        return self._thread_local.connection

    def close(self):
        """關閉當前線程的連接"""
        if hasattr(self._thread_local, 'connection'):
            self._thread_local.connection.close()
            del self._thread_local.connection


# 使用示例
from django.db import connection

def view_function(request):
    # 線程 A 執行查詢
    users = User.objects.all()  # 使用線程 A 的連接

    # 線程 B 同時執行查詢
    # 會自動使用線程 B 的連接，不會衝突！

# django/db/transaction.py (簡化)

class Atomic:
    def __init__(self, using=None, savepoint=True):
        self.using = using
        self.savepoint = savepoint

    def __enter__(self):
        connection = get_connection(self.using)

        # 檢查當前線程的事務狀態
        if not connection.in_atomic_block:
            # 開始新事務
            connection.set_autocommit(False)

        # 保存當前線程的事務狀態
        connection.in_atomic_block = True
        connection.savepoint_ids.append(
            connection.savepoint() if self.savepoint else None
        )

    def __exit__(self, exc_type, exc_value, traceback):
        connection = get_connection(self.using)

        if exc_type is None:
            # 提交當前線程的事務
            connection.commit()
        else:
            # 回滾當前線程的事務
            connection.rollback()


# 使用示例
from django.db import transaction

def transfer_money(from_user, to_user, amount):
    with transaction.atomic():
        # 這個事務只影響當前線程
        from_user.balance -= amount
        from_user.save()

        to_user.balance += amount
        to_user.save()

        # 其他線程的事務不受影響

import threading

# 全局的 thread local 對象
_thread_locals = threading.local()

class ThreadLocalMiddleware:
    """在 threading.local 中存儲請求對象"""

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

    def __call__(self, request):
        # 將請求存儲到當前線程
        _thread_locals.request = request
        _thread_locals.user = request.user

        response = self.get_response(request)

        # 清理（可選）
        if hasattr(_thread_locals, 'request'):
            del _thread_locals.request
        if hasattr(_thread_locals, 'user'):
            del _thread_locals.user

        return response


def get_current_user():
    """在任何地方獲取當前用戶"""
    return getattr(_thread_locals, 'user', None)


# 在視圖、模型、工具函數中都可以使用
def some_utility_function():
    user = get_current_user()
    if user and user.is_authenticated:
        log.info(f"用戶 {user.username} 執行了操作")

# settings.py
MIDDLEWARE = [
    'myapp.middleware.ThreadLocalMiddleware',  # 添加自定義中間件
    'django.middleware.security.SecurityMiddleware',
    # ...
]

import threading
import logging

# 創建 thread local 對象
thread_local = threading.local()

class RequestIdFilter(logging.Filter):
    """在日誌中添加請求 ID"""

    def filter(self, record):
        record.request_id = getattr(thread_local, 'request_id', 'N/A')
        return True


# 配置日誌
logging.basicConfig(
    format='%(asctime)s [%(request_id)s] %(message)s',
    level=logging.INFO
)
logger = logging.getLogger()
logger.addFilter(RequestIdFilter())


# 在中間件中設置請求 ID
class RequestIdMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        import uuid
        thread_local.request_id = str(uuid.uuid4())[:8]

        response = self.get_response(request)
        return response


# 在任何地方記錄日誌都會帶上請求 ID
def my_view(request):
    logger.info("處理用戶請求")  # 輸出: 2025-01-14 11:40:00 [a3b4c5d6] 處理用戶請求
    return HttpResponse("OK")

import threading
from concurrent.futures import ThreadPoolExecutor

thread_local = threading.local()

def worker(task_id):
    # ⚠️ 線程池中的線程會被復用！
    if not hasattr(thread_local, 'counter'):
        thread_local.counter = 0

    thread_local.counter += 1
    print(f"任務 {task_id}: counter = {thread_local.counter}")


# 使用線程池（線程會被復用）
with ThreadPoolExecutor(max_workers=2) as executor:
    for i in range(6):
        executor.submit(worker, i)

# 可能的輸出：
# 任務 0: counter = 1
# 任務 1: counter = 1
# 任務 2: counter = 2  # 線程被復用，counter 累加！
# 任務 3: counter = 2
# 任務 4: counter = 3  # 繼續累加
# 任務 5: counter = 3

def worker(task_id):
    # ✅ 顯式重置
    thread_local.counter = 0

    thread_local.counter += 1
    print(f"任務 {task_id}: counter = {thread_local.counter}")

import threading

thread_local = threading.local()

def create_many_threads():
    threads = []
    for i in range(1000):
        t = threading.Thread(target=lambda: setattr(thread_local, 'data', [0] * 1000000))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

    # ⚠️ 線程結束後，thread_local 的資料應該被清理
    # 但如果線程對象還被引用，資料不會被釋放！


# ✅ 正確做法：不持有線程對象的引用
def create_many_threads_correct():
    for i in range(1000):
        t = threading.Thread(target=lambda: setattr(thread_local, 'data', [0] * 1000000))
        t.start()
        t.join()  # 等待線程結束，然後立即釋放

import asyncio
import threading

thread_local = threading.local()

async def async_worker(name):
    # ❌ 在 asyncio 中，多個協程可能在同一個線程中執行
    thread_local.user = name
    await asyncio.sleep(0.1)  # 切換協程

    print(f"{name} 看到的用戶: {thread_local.user}")
    # 可能被其他協程修改！


# 運行
asyncio.run(asyncio.gather(
    async_worker("Alice"),
    async_worker("Bob"),
))

# 可能的輸出：
# Alice 看到的用戶: Bob  # 錯誤！
# Bob 看到的用戶: Bob


# ✅ 正確做法：使用 contextvars
from contextvars import ContextVar

user_var = ContextVar('user', default=None)

async def async_worker_correct(name):
    user_var.set(name)
    await asyncio.sleep(0.1)

    print(f"{name} 看到的用戶: {user_var.get()}")
    # 正確！每個協程有自己的上下文


asyncio.run(asyncio.gather(
    async_worker_correct("Alice"),
    async_worker_correct("Bob"),
))

# 輸出：
# Alice 看到的用戶: Alice
# Bob 看到的用戶: Bob

# 1. 如果使用傳統多線程（WSGI、threading）
import threading
thread_local = threading.local()

# 2. 如果使用 asyncio、ASGI（FastAPI、Starlette）
from contextvars import ContextVar
context_var = ContextVar('name')

# 3. 如果調用鏈很短（< 3 層）
def view(request):
    process_data(request.user)  # 直接傳遞

# 4. 如果需要跨線程共享
from threading import Lock
shared_data = {}
shared_lock = Lock()

# middleware/request_tracking.py

import threading
import uuid
import logging
import time

# 創建 thread local 對象
_request_context = threading.local()


class RequestTrackingMiddleware:
    """請求追蹤中間件"""

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

    def __call__(self, request):
        # 生成請求 ID
        request_id = str(uuid.uuid4())
        request.id = request_id

        # 存儲到 thread local
        _request_context.request_id = request_id
        _request_context.user_id = request.user.id if request.user.is_authenticated else None
        _request_context.path = request.path
        _request_context.method = request.method
        _request_context.start_time = time.time()

        # 處理請求
        response = self.get_response(request)

        # 計算耗時
        duration = time.time() - _request_context.start_time

        # 記錄請求日誌
        logger.info(
            f"{request.method} {request.path} "
            f"[{response.status_code}] "
            f"{duration*1000:.2f}ms"
        )

        # 清理上下文
        self._clear_context()

        return response

    def _clear_context(self):
        """清理線程本地資料"""
        for attr in ['request_id', 'user_id', 'path', 'method', 'start_time']:
            if hasattr(_request_context, attr):
                delattr(_request_context, attr)


# 自定義日誌過濾器
class RequestContextFilter(logging.Filter):
    """在日誌中添加請求上下文"""

    def filter(self, record):
        # 從 thread local 獲取資訊
        record.request_id = getattr(_request_context, 'request_id', '-')
        record.user_id = getattr(_request_context, 'user_id', '-')
        return True


# 配置日誌
logger = logging.getLogger('django')
logger.addFilter(RequestContextFilter())


# 工具函數：獲取當前請求 ID
def get_current_request_id():
    """在任何地方獲取當前請求 ID"""
    return getattr(_request_context, 'request_id', None)


def get_current_user_id():
    """在任何地方獲取當前用戶 ID"""
    return getattr(_request_context, 'user_id', None)


# 使用示例
def some_service_function():
    """在業務邏輯中使用請求上下文"""
    request_id = get_current_request_id()
    user_id = get_current_user_id()

    logger.info(f"執行業務邏輯")
    # 日誌輸出: [a3b4c5d6] [user:123] 執行業務邏輯

    # 調用外部 API 時傳遞 request_id
    response = requests.post(
        'https://api.example.com/endpoint',
        headers={'X-Request-ID': request_id}
    )

# settings.py

MIDDLEWARE = [
    'myapp.middleware.request_tracking.RequestTrackingMiddleware',  # 第一個
    'django.middleware.security.SecurityMiddleware',
    # ...
]

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s [%(request_id)s] [user:%(user_id)s] %(levelname)s %(message)s'
        },
    },
    'filters': {
        'request_context': {
            '()': 'myapp.middleware.request_tracking.RequestContextFilter',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'detailed',
            'filters': ['request_context'],
        },
    },
    'root': {
        'handlers': ['console'],
        'level': 'INFO',
    },
}

# 全局變量 - 所有線程共享
count = 0

def worker():
    global count
    for _ in range(10000):
        count += 1  # 需要加鎖！


# threading.local - 每個線程獨立
thread_local = threading.local()

def worker():
    thread_local.count = 0
    for _ in range(10000):
        thread_local.count += 1  # 不需要加鎖

# 如果不用 threading.local（錯誤示例）
global_connection = create_connection()

def view1(request):
    global_connection.execute("BEGIN")  # 線程 A 開始事務
    # ...

def view2(request):
    global_connection.execute("BEGIN")  # 線程 B 開始事務
    # ❌ 衝突！兩個事務共用一個連接


# 使用 threading.local（正確）
class DatabaseWrapper:
    def __init__(self):
        self._local = threading.local()

    @property
    def connection(self):
        if not hasattr(self._local, 'conn'):
            self._local.conn = create_connection()  # 每個線程獨立連接
        return self._local.conn

import asyncio
import threading

thread_local = threading.local()

async def task(name):
    thread_local.name = name
    await asyncio.sleep(0.1)  # 切換協程
    print(f"{name} 看到的 name: {thread_local.name}")  # 可能被覆蓋


asyncio.run(asyncio.gather(task("A"), task("B")))
# 輸出: A 看到的 name: B  ❌ 錯誤！

from contextvars import ContextVar

name_var = ContextVar('name')

async def task(name):
    name_var.set(name)
    await asyncio.sleep(0.1)
    print(f"{name} 看到的 name: {name_var.get()}")  # 正確


asyncio.run(asyncio.gather(task("A"), task("B")))
# 輸出:
# A 看到的 name: A  ✅
# B 看到的 name: B  ✅

import threading
import tracemalloc

thread_local = threading.local()

def worker():
    # 存儲大量資料
    thread_local.data = [0] * 1000000  # 分配 1MB
    # 線程結束


# 開始追蹤
tracemalloc.start()

# 創建大量線程
threads = []
for i in range(100):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)  # ❌ 持有線程引用！

for t in threads:
    t.join()

# 查看記憶體使用
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

for stat in top_stats[:10]:
    print(stat)

# 輸出：可能顯示 100MB 未釋放（100 個線程 × 1MB）


# ✅ 修復：不持有線程引用
for i in range(100):
    t = threading.Thread(target=worker)
    t.start()
    t.join()  # 立即等待並釋放

# 記憶體正常釋放

import threading

# ❌ 這樣不行
thread_local = threading.local()
thread_local.count = 0  # 只在當前線程有效！


# ✅ 正確方法：繼承 threading.local
class MyLocal(threading.local):
    def __init__(self):
        # 每個線程第一次訪問時都會調用
        self.count = 0
        self.name = "default"


my_local = MyLocal()

def worker(id):
    print(f"線程 {id}: count = {my_local.count}")  # 都是 0
    my_local.count += 1
    print(f"線程 {id}: count = {my_local.count}")  # 都是 1


import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(worker, [1, 2, 3])

# 輸出：
# 線程 1: count = 0
# 線程 1: count = 1
# 線程 2: count = 0
# 線程 2: count = 1
# 線程 3: count = 0
# 線程 3: count = 1