Python使用threading.local()实现线程局部存储的完全指南

import threading

tls = threading.local()   # 1. 创建 TLS 对象

def worker(idx):
    tls.value = idx       # 2. 各线程写自己的值
    print(f'Thread {idx} sees {tls.value}')

for i in range(5):
    threading.Thread(target=worker, args=(i,)).start()

Thread 0 sees 0
Thread 4 sees 4
Thread 1 sees 1
Thread 2 sees 2
Thread 3 sees 3

# 伪代码
current_thread_dict[id(tls)]['xxx'] = value

import threading
import time

_ctx = threading.local()

def api_handler(request_id):
    _ctx.request_id = request_id
    business_logic()

def business_logic():
    # 任意深处都能拿到 request_id，而不用层层传参
    print(f'Handling {threading.current_thread().name}  req={_ctx.request_id}')
    time.sleep(0.1)

for rid in range(3):
    threading.Thread(target=api_handler, args=(rid,), name=f'T{rid}').start()

import sqlite3, threading

db_local = threading.local()

def get_conn():
    """每个线程首次调用时创建连接，后续复用"""
    if not hasattr(db_local, 'conn'):
        db_local.conn = sqlite3.connect(':memory:')
    return db_local.conn

def worker():
    conn = get_conn()
    conn.execute('create table if not exists t(x)')
    conn.execute('insert into t values (1)')
    conn.commit()
    print(f'{threading.current_thread().name}  inserted')

threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()

import threading, random, time

local = threading.local()

def job():
    local.val = random.randint(1, 100)
    time.sleep(0.1)
    assert local.val == threading.local().val, "Should never fail!"
    print(f'{threading.current_thread().name}  val={local.val}')

for _ in range(10):
    threading.Thread(target=job).start()