SERIALIZABLE 实现
│
├── 1. 快照读（普通 SELECT）
│   └── 自动转为 LOCK IN SHARE MODE（共享读锁）
│
├── 2. 当前读（FOR UPDATE / UPDATE / DELETE）
│   └── 与 REPEATABLE READ 一致，使用 Next-Key Lock
│
└── 3. 自动加锁规则
    └── 所有读操作都加锁，读写互斥

-- SERIALIZABLE 下
BEGIN;
SELECT * FROM user WHERE id = 1;
-- 等价于：SELECT * FROM user WHERE id = 1 LOCK IN SHARE MODE;
-- 其他事务无法对 id=1 的行加排他锁（不能修改）

SELECT * FROM user WHERE age > 20 FOR UPDATE;
-- SERIALIZABLE 级别：Next-Key Lock 锁定范围

-- 1. 特定事务临时提升隔离级别
SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
BEGIN;
-- 执行需要最高一致性的操作
COMMIT;
SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;

READ COMMITTED  ：无防幻读能力
REPEATABLE READ：MVCC（快照读）+ Gap Lock（当前读）
SERIALIZABLE   ：所有读加共享锁 + Gap Lock（强制串行化）

graph LR
    A[长连接建立] --> B[执行SQL]
    B --> C[临时内存分配<br/>会话级别]
    C --> D[持续累积]
    D --> E[内存溢出<br/>MySQL被OOM Killer杀掉]

-- 每次执行都会创建临时表，连接不释放就不销毁
SELECT * FROM t1 JOIN t2 ON t1.id = t2.id ORDER BY t1.name;

-- 预编译的SQL会占用连接内存
PREPARE stmt FROM 'SELECT * FROM user WHERE id = ?';
EXECUTE stmt USING @id;

flowchart TD
    subgraph 方案A[定期断开重连]
        A1[执行N条SQL后<br/>主动断开连接]
        A2[释放所有会话内存]
        A3[重新建立新连接]
        A1 --> A2 --> A3
    end

    subgraph 方案B[连接池+reset]
        B1[连接归还池前<br/>执行mysql_reset_connection]
        B2[清理会话状态<br/>释放临时内存]
        B3[连接池复用]
        B1 --> B2 --> B3
    end

    subgraph 方案C[设置超时]
        C1[wait_timeout<br/>缩小空闲超时]
        C2[interactive_timeout<br/>交互式超时]
        C1 --> C2
    end

# 伪代码示例
MAX_QUERIES = 10000
query_count = 0

while True:
    cursor.execute(sql)
    query_count += 1
    if query_count >= MAX_QUERIES:
        cursor.close()          # 断开连接，释放内存
        cursor = connect()      # 重连
        query_count = 0

-- 连接复用前重置会话状态
mysql_reset_connection(conn);  -- C API

[mysqld]
# 非交互式连接超时时间（默认28800秒=8小时）
wait_timeout = 600

# 交互式连接超时时间
interactive_timeout = 600

# 最大连接数（防止连接过多撑爆内存）
max_connections = 500

应用 -> 连接池(Druid/HikariCP) -> MySQL

-- 查看哪个连接占用内存较大
SELECT * FROM sys.session ORDER BY (current_memory) DESC LIMIT 5;

-- 查看连接数分布
SELECT * FROM information_schema.PROCESSLIST;

graph LR
    A[长连接建立] --> B[执行SQL]
    B --> C[临时内存分配<br/>会话级别]
    C --> D[持续累积]
    D --> E[内存溢出<br/>MySQL被OOM Killer杀掉]

-- 每次执行都会创建临时表，连接不释放就不销毁
SELECT * FROM t1 JOIN t2 ON t1.id = t2.id ORDER BY t1.name;

-- 预编译的SQL会占用连接内存
PREPARE stmt FROM 'SELECT * FROM user WHERE id = ?';
EXECUTE stmt USING @id;

flowchart TD
    subgraph 方案A[定期断开重连]
        A1[执行N条SQL后<br/>主动断开连接]
        A2[释放所有会话内存]
        A3[重新建立新连接]
        A1 --> A2 --> A3
    end

    subgraph 方案B[连接池+reset]
        B1[连接归还池前<br/>执行mysql_reset_connection]
        B2[清理会话状态<br/>释放临时内存]
        B3[连接池复用]
        B1 --> B2 --> B3
    end

    subgraph 方案C[设置超时]
        C1[wait_timeout<br/>缩小空闲超时]
        C2[interactive_timeout<br/>交互式超时]
        C1 --> C2
    end

# 伪代码示例
MAX_QUERIES = 10000
query_count = 0

while True:
    cursor.execute(sql)
    query_count += 1
    if query_count >= MAX_QUERIES:
        cursor.close()          # 断开连接，释放内存
        cursor = connect()      # 重连
        query_count = 0

-- 连接复用前重置会话状态
mysql_reset_connection(conn);  -- C API

[mysqld]
# 非交互式连接超时时间（默认28800秒=8小时）
wait_timeout = 600

# 交互式连接超时时间
interactive_timeout = 600

# 最大连接数（防止连接过多撑爆内存）
max_connections = 500

应用 -> 连接池(Druid/HikariCP) -> MySQL

-- 查看哪个连接占用内存较大
SELECT * FROM sys.session ORDER BY (current_memory) DESC LIMIT 5;

-- 查看连接数分布
SELECT * FROM information_schema.PROCESSLIST;

import weakref
import gc

class MyObject:
    def __init__(self, name):
        self.name = name
        print(f"{name} 创建")

    def __del__(self):
        print(f"{self.name} 销毁")

# 强引用
obj = MyObject("测试对象")  # 引用计数 = 1

# 弱引用
weak = weakref.ref(obj)
print(weak())  # <__main__.MyObject object at ...> — 可以访问

del obj  # 引用计数归零，对象销毁
print(weak())  # None — 对象已被回收

import weakref

class Data:
    def __init__(self, value):
        self.value = value

data = Data(42)

# 创建弱引用
ref = weakref.ref(data)
print(ref())      # <__main__.Data object at ...>
print(ref().value)  # 42

# 获取引用的回调通知
def on_delete(ref):
    print(f"对象已被回收")

ref_with_callback = weakref.ref(data, on_delete)

del data  # 触发 on_delete 回调

from weakref import WeakValueDictionary

class Cache:
    def __init__(self):
        self._cache = WeakValueDictionary()  # 值用弱引用

    def add(self, key, value):
        self._cache[key] = value

    def get(self, key):
        return self._cache.get(key)

cache = Cache()
obj = MyObject("缓存对象")
cache.add("key1", obj)

print(cache.get("key1"))  # <...> — 存在

del obj  # 对象被回收
print(cache.get("key1"))  # None — 自动消失，没有内存泄漏

from weakref import WeakSet

class EventHandler:
    def __init__(self, name):
        self.name = name

    def handle(self, event):
        print(f"{self.name} 处理了 {event}")

class EventManager:
    def __init__(self):
        self._handlers = WeakSet()  # 自动排除已销毁的 handler

    def register(self, handler):
        self._handlers.add(handler)

    def fire(self, event):
        for handler in self._handlers:
            handler.handle(event)

manager = EventManager()
handler = EventHandler("处理器A")
manager.register(handler)

manager.fire("事件1")  # 处理器A 处理了 事件1

del handler  # 销毁处理器
manager.fire("事件2")  # 没有任何输出

# 1. 不能弱引用基本类型
import weakref
# weakref.ref(42)  # ❌ TypeError: cannot create weak reference to 'int' object

# 2. 不能弱引用 str
# weakref.ref("hello")  # ❌ TypeError

# 3. 不能弱引用 list、dict、tuple
# weakref.ref([1, 2, 3])  # ❌ TypeError

# 但可以通过继承支持
class MyList(list):
    pass  # 自定义的 list 子类支持弱引用

my_list = MyList([1, 2, 3])
ref = weakref.ref(my_list)
print(ref())  # [1, 2, 3]

class MyClass:
    __slots__ = ('data', '__weakref__')  # 显式包含 __weakref__

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

# __slots__ 默认不包含 __weakref__，需要显式包含

from weakref import WeakSet

class Observable:
    def __init__(self):
        self._observers = WeakSet()

    def attach(self, observer):
        self._observers.add(observer)

    def detach(self, observer):
        self._observers.discard(observer)

    def notify(self, message):
        for observer in self._observers:
            observer.update(message)

class Observer:
    def __init__(self, name):
        self.name = name

    def update(self, message):
        print(f"{self.name} 收到: {message}")

    def __del__(self):
        print(f"{self.name} 被销毁")

subject = Observable()
obs1 = Observer("观察者1")
obs2 = Observer("观察者2")

subject.attach(obs1)
subject.attach(obs2)

subject.notify("你好！")
del obs1  # 观察者1 被销毁
# WeakSet 自动移除失效引用
subject.notify("再见！")  # 只有观察者2 收到

import weakref

class ImageLoader:
    _cache = weakref.WeakValueDictionary()

    @classmethod
    def load(cls, path):
        img = cls._cache.get(path)
        if img is None:
            img = cls._load_from_disk(path)
            cls._cache[path] = img
        return img

    @classmethod
    def _load_from_disk(cls, path):
        print(f"从磁盘加载: {path}")
        return Image(path)

class Image:
    def __init__(self, path):
        self.path = path
        self.data = f"图片数据:{path}"

    def __del__(self):
        print(f"图片 {self.path} 释放")

# 使用
img1 = ImageLoader.load("photo.jpg")
img2 = ImageLoader.load("photo.jpg")  # 从缓存获取
print(img1 is img2)  # True

del img1
del img2  # 所有强引用消失，图片从缓存自动移除

# ❌ 循环引用
class Parent:
    def __init__(self):
        self.child = None

class Child:
    def __init__(self, parent):
        self.parent = parent  # 强引用

p = Parent()
c = Child(p)
p.child = c  # 双向引用循环

# ✅ 用弱引用破环
class Parent:
    def __init__(self):
        self.child = None

class Child:
    def __init__(self, parent):
        self.parent = weakref.ref(parent)  # 弱引用

p = Parent()
c = Child(p)
p.child = c

print(c.parent())  # 
del p
print(c.parent())  # None — 不会阻止回收

import weakref

# 四种 weakref 容器
ref = weakref.ref(obj)              # 单个弱引用
proxy = weakref.proxy(obj)          # 弱引用代理（透明访问）
dict = weakref.WeakValueDictionary() # 值的弱引用字典
set = weakref.WeakSet()             # 弱引用集合

class Node:
    def __init__(self, name):
        self.name = name
        self.parent = None
        self.children = []

def create_cycle():
    a = Node("A")
    b = Node("B")
    a.children.append(b)  # a → b
    b.parent = a          # b → a，形成循环

create_cycle()
# a 和 b 在函数结束后互相引用，引用计数无法归零

graph LR
    A[对象 A ref=2] -->|children| B[对象 B ref=2]
    B -->|parent| A

    style A fill:#ff9999
    style B fill:#ff9999

# 列表循环引用
a = []
a.append(a)  # 列表引用自身

# 字典循环引用
d = {}
d["self"] = d

# 函数闭包的循环引用
def outer():
    x = []
    def inner():
        return x
    return inner

import gc
import sys

class LeakDemo:
    def __init__(self):
        self.ref = None

# 故意制造泄漏
leaks = []
for i in range(10000):
    a = LeakDemo()
    b = LeakDemo()
    a.ref = b
    b.ref = a
    leaks.append(a)

print("GC未回收前:", len(gc.get_objects()))
# 手动触发回收
n = gc.collect()
print(f"回收了 {n} 个对象")

import gc
import objgraph  # 第三方库

# 查看对象的引用链
gc.collect()
objgraph.show_backrefs(
    [leaks[0]], 
    max_depth=5, 
    filename="cycle.png"
)

import weakref

class Node:
    def __init__(self, name):
        self.name = name
        self._parent = None

    @property
    def parent(self):
        return self._parent() if self._parent else None

    @parent.setter
    def parent(self, node):
        self._parent = weakref.ref(node) if node else None

a = Node("A")
b = Node("B")
b.parent = a       # 弱引用，不增加引用计数
a.children.append(b)  # 普通引用
# 函数结束后可以被正常回收

import gc

# 查找无法回收的循环引用对象
gc.set_debug(gc.DEBUG_SAVEALL)

# 手动触发回收
unreachable = gc.collect()
print(f"不可达对象数量: {unreachable}")

# 检查对象是否被追踪
print(gc.is_tracked(a))

class Danger:
    def __del__(self):
        print("Danger 被清理")

a = Danger()
b = Danger()
a.ref = b
b.ref = a
del a, b
# GC 无法回收 Danger 实例，__del__ 永远不会被调用！

from abc import ABC, abstractmethod

# 抽象产品
class Database(ABC):
    @abstractmethod
    def connect(self):
        pass

# 具体产品
class MySQL(Database):
    def connect(self):
        return "已连接到 MySQL"

class PostgreSQL(Database):
    def connect(self):
        return "已连接到 PostgreSQL"

class SQLite(Database):
    def connect(self):
        return "已连接到 SQLite"

# 简单工厂
class DatabaseFactory:
    @staticmethod
    def create(db_type: str) -> Database:
        factories = {
            "mysql": MySQL,
            "postgres": PostgreSQL,
            "sqlite": SQLite,
        }
        cls = factories.get(db_type.lower())
        if cls is None:
            raise ValueError(f"不支持的数据库类型: {db_type}")
        return cls()

# 使用
db = DatabaseFactory.create("mysql")
print(db.connect())  # 已连接到 MySQL

from abc import ABC, abstractmethod

class Document(ABC):
    @abstractmethod
    def open(self):
        pass

    @abstractmethod
    def save(self, content):
        pass

class PDF(Document):
    def open(self):
        return "打开 PDF 文档"

    def save(self, content):
        return f"保存 PDF: {content}"

class Word(Document):
    def open(self):
        return "打开 Word 文档"

    def save(self, content):
        return f"保存 Word: {content}"

# 创建器
class DocumentCreator(ABC):
    @abstractmethod
    def create_document(self) -> Document:
        pass

    def edit_document(self):
        doc = self.create_document()
        doc.open()
        doc.save("内容")

class PDFCreator(DocumentCreator):
    def create_document(self) -> Document:
        return PDF()

class WordCreator(DocumentCreator):
    def create_document(self) -> Document:
        return Word()

# 使用
creator = PDFCreator()
creator.edit_document()  # 使用 PDF

classDiagram
    class Creator {
        +factory_method() Document
        +operation()
    }
    class ConcreteCreatorA {
        +factory_method() Document
    }
    class ConcreteCreatorB {
        +factory_method() Document
    }
    class Document {
        <<abstract>>
    }
    class ProductA
    class ProductB

    Creator <|-- ConcreteCreatorA
    Creator <|-- ConcreteCreatorB
    ConcreteCreatorA --> ProductA
    ConcreteCreatorB --> ProductB
    Document <|-- ProductA
    Document <|-- ProductB

from abc import ABC, abstractmethod

# 抽象产品族
class Button(ABC):
    @abstractmethod
    def render(self):
        pass

class Checkbox(ABC):
    @abstractmethod
    def render(self):
        pass

# 具体产品
class WindowsButton(Button):
    def render(self):
        return "Windows 风格的按钮"

class WindowsCheckbox(Checkbox):
    def render(self):
        return "Windows 风格的复选框"

class MacButton(Button):
    def render(self):
        return "Mac 风格的按钮"

class MacCheckbox(Checkbox):
    def render(self):
        return "Mac 风格的复选框"

# 抽象工厂
class UIFactory(ABC):
    @abstractmethod
    def create_button(self) -> Button:
        pass

    @abstractmethod
    def create_checkbox(self) -> Checkbox:
        pass

class WindowsFactory(UIFactory):
    def create_button(self) -> Button:
        return WindowsButton()
    def create_checkbox(self) -> Checkbox:
        return WindowsCheckbox()

class MacFactory(UIFactory):
    def create_button(self) -> Button:
        return MacButton()
    def create_checkbox(self) -> Checkbox:
        return MacCheckbox()

# 使用
factory = WindowsFactory()  # 或 MacFactory()
button = factory.create_button()
checkbox = factory.create_checkbox()

import logging
from abc import ABC, abstractmethod

class Logger(ABC):
    @abstractmethod
    def log(self, message: str):
        pass

class FileLogger(Logger):
    def __init__(self, path):
        self.path = path

    def log(self, message: str):
        with open(self.path, "a") as f:
            f.write(f"{message}\n")

class ConsoleLogger(Logger):
    def log(self, message: str):
        print(f"[LOG] {message}")

class CloudLogger(Logger):
    def log(self, message: str):
        import requests
        requests.post("https://log-service/api/log", json={"msg": message})

class LoggerFactory:
    _loggers = {
        "file": FileLogger,
        "console": ConsoleLogger,
        "cloud": CloudLogger,
    }

    @classmethod
    def register(cls, name, logger_cls):
        cls._loggers[name] = logger_cls

    @classmethod
    def create(cls, name: str, **kwargs) -> Logger:
        logger_cls = cls._loggers.get(name)
        if not logger_cls:
            raise ValueError(f"unknown logger: {name}")
        return logger_cls(**kwargs)

# 注册自定义日志器
LoggerFactory.register("redis", RedisLogger)

# 使用
logger = LoggerFactory.create("file", path="/var/log/app.log")
logger.log("系统启动")

func TestA(t *testing.T) {
    t.Parallel() // 标记为可并行
    // 测试逻辑
}

func TestB(t *testing.T) {
    t.Parallel()
    // 测试逻辑
}

// 当 TestA 和 TestB 都调用 t.Parallel()，
// 它们可能被同时执行

sequenceDiagram
    participant Runner as 测试运行器
    participant TestA
    participant TestB
    participant Resource as 共享资源

    Note over Runner: 启动测试
    Runner->>TestA: t.Parallel()
    Note over TestA: 暂停等待
    Runner->>TestB: t.Parallel()
    Note over TestB: 暂停等待
    Note over Runner: 所有 t.Parallel() 测试<br/>同时恢复
    par 并行执行
        TestA->>Resource: 写入数据
        TestB->>Resource: 读取数据
    end
    Note over Resource: ⚠️ 数据竞争！

var counter int

// ❌ 并行测试共享 counter
func TestCounterA(t *testing.T) {
    t.Parallel()
    for i := 0; i < 1000; i++ {
        counter++ // 数据竞争
    }
}

func TestCounterB(t *testing.T) {
    t.Parallel()
    for i := 0; i < 1000; i++ {
        counter++ // 数据竞争
    }
}

// ❌ 并行测试使用同一个临时文件
func TestWriteFile(t *testing.T) {
    t.Parallel()
    tmpFile := "/tmp/test.txt"
    os.WriteFile(tmpFile, []byte("data1"), 0644)
    // 数据可能被另一个测试覆盖
    data, _ := os.ReadFile(tmpFile)
    // 读取的内容不确定
}

// ❌ 并行测试修改同一张表
func TestInsertUser(t *testing.T) {
    t.Parallel()
    db.Exec("INSERT INTO users(name) VALUES('alice')")
}

func TestDeleteUser(t *testing.T) {
    t.Parallel()
    db.Exec("DELETE FROM users WHERE name='alice'")
    // 可能删除 TestInsertUser 刚插入的数据
}

func TestParallelLog(t *testing.T) {
    t.Parallel()
    for i := 0; i < 100; i++ {
        go func(n int) {
            t.Logf("iteration %d", n) // t.Logf 不是 100% 无竞态的
        }(i)
    }
}

func TestParallel(t *testing.T) {
    // 每个子测试可以有独立的 setup/teardown
    tests := []struct {
        name string
        data string
    }{
        {"case1", "data1"},
        {"case2", "data2"},
    }

    for _, tc := range tests {
        tc := tc // 捕获循环变量
        t.Run(tc.name, func(t *testing.T) {
            t.Parallel()
            // 每个子测试独立运行
            fmt.Println(tc.data)
        })
    }
}

func TestWriteFile(t *testing.T) {
    t.Parallel()

    // ✅ 每个测试创建自己的临时目录
    tmpDir := t.TempDir()
    tmpFile := filepath.Join(tmpDir, "test.txt")

    os.WriteFile(tmpFile, []byte("data1"), 0644)
    data, _ := os.ReadFile(tmpFile)
    // 不会被其他并行测试影响
}

func TestDatabase(t *testing.T) {
    t.Parallel()

    // ✅ 每个测试在自己的事务中运行
    tx := db.Begin()
    defer tx.Rollback() // 测试结束后回滚

    _, err := tx.Exec("INSERT INTO users(name) VALUES('alice')")
    // 其他并行测试看不到这个数据
}

var (
    sharedData map[string]string
    dataMu     sync.RWMutex
)

func TestSharedData(t *testing.T) {
    t.Parallel()

    // ✅ 加锁保护
    dataMu.Lock()
    sharedData["key"] = "value"
    dataMu.Unlock()

    dataMu.RLock()
    v := sharedData["key"]
    dataMu.RUnlock()
}

# 启用 race detector 运行测试
go test -race ./...

# 并行运行测试
go test -parallel 4 -race ./...

flowchart TD
    A["Setup (串行)"] --> B["t.Parallel()"]
    B --> C1["测试数据/配置初始化"]
    B --> C2["测试数据/配置初始化"]
    B --> C3["测试数据/配置初始化"]
    C1 --> D1["并行测试 A"]
    C2 --> D2["并行测试 B"]
    C3 --> D3["并行测试 C"]
    D1 --> E["Cleanup (串行)"]
    D2 --> E
    D3 --> E

func TestBestPractices(t *testing.T) {
    // 1. 串行初始化
    globalCfg := loadConfig()

    tests := []struct {
        name string
        fn   func(*testing.T, Config)
    }{
        {"test-a", testA},
        {"test-b", testB},
    }

    for _, tt := range tests {
        tt := tt
        t.Run(tt.name, func(t *testing.T) {
            t.Parallel() // 2. 子测试中调用

            // 3. 每个子测试用独立资源
            cfg := globalCfg.Copy()
            tmpDir := t.TempDir()

            // 4. 执行测试
            tt.fn(t, cfg)
        })
    }
}

// time.After 返回一个通道，在指定时间后接收到一个 Time 值
func After(d Duration) <-chan Time {
    return NewTimer(d).C
}

// ❌ 高频循环中使用 time.After 会泄漏！
func leakInLoop() {
    for {
        select {
        case msg := <-ch:
            process(msg) // 假设 msg 来得很频繁
        case <-time.After(5 * time.Minute):
            // 5 分钟后超时处理
            fmt.Println("超时")
        }
    }
}

sequenceDiagram
    participant L as 循环
    participant TA as time.After
    participant T as Timer
    participant G as GC

    L->>TA: 第 1 次 time.After(5min)
    TA->>T: 创建 Timer（5 分钟后触发）
    Note over L:  msg 很快就来了
    L->>T: select 选择了 msg，Timer 被丢弃
    Note over T,G: Timer 还活 5 分钟<br>GC 无法回收<br>（因为被定时器堆引用）

    L->>TA: 第 2 次 time.After(5min)
    TA->>T: 又创建一个 Timer
    L->>T: 又丢弃...
    Note over T: 第 2 个 Timer 泄漏

    Note over L: 循环 100 万次 → 100 万个 Timer 泄漏

// 每次泄漏的 Timer 大约占 200 字节
// 循环 100 万次 ≈ 泄漏 200MB 内存
// 而且每个 Timer 绑定了一个 goroutine 等待触发

// ✅ 手动创建并复用 Timer
func loopWithTimer() {
    timer := time.NewTimer(5 * time.Minute)
    defer timer.Stop()

    for {
        // 先将 timer 暂停
        timer.Stop()
        // 清理通道（如果已触发）
        select {
        case <-timer.C:
        default:
        }
        timer.Reset(5 * time.Minute)

        select {
        case msg := <-ch:
            process(msg)
        case <-timer.C:
            fmt.Println("超时")
        }
    }
}

// ✅ 在一个循环周期内只创建一个 time.After
func loopWithAfter() {
    timeout := time.After(5 * time.Minute)

    for {
        select {
        case msg := <-ch:
            process(msg)
        case <-timeout:
            fmt.Println("超时，退出循环")
            return
        }
    }
    // 注意：这适用于一次超时就退出的场景
}

// ✅ 使用 context 代替 time.After
func loopWithContext(ctx context.Context) {
    for {
        select {
        case msg := <-ch:
            process(msg)
        case <-ctx.Done():
            fmt.Println("超时或取消")
            return
        }
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    loopWithContext(ctx)
}

// time.Tick 也有同样的问题——无法停止
func tickLeak() {
    for range time.Tick(time.Second) {
        // time.Tick 创建的 Ticker 无法停止
        // 即使循环退出了，Ticker 仍然运行
    }
}

// ✅ 使用 NewTicker
func tickSafe() {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for range ticker.C {
        // ...
    }
}

// ✅ 一次性超时——安全
func doOnce() error {
    done := make(chan error, 1)
    go func() {
        done <- doWork()
    }()

    // 只执行一次，Timer 不是"累积"的，安全
    select {
    case err := <-done:
        return err
    case <-time.After(30 * time.Second):
        return fmt.Errorf("超时")
    }
}

// ✅ 函数退出时 select 也退出——安全
func safeSelect() {
    ch := make(chan int)

    // 函数返回时 select 退出
    // 虽然仍有一个 Timer 跑着，但函数不循环，影响有限
    select {
    case v := <-ch:
        use(v)
    case <-time.After(time.Second):
        fmt.Println("超时")
    }
}

func detectTimerLeak() {
    var before, after runtime.MemStats

    runtime.ReadMemStats(&before)

    // 模拟泄漏
    for i := 0; i < 10000; i++ {
        go leakInLoop()
    }

    runtime.ReadMemStats(&after)
    fmt.Printf("内存增长: %d MB\n", (after.HeapAlloc-before.HeapAlloc)/1024/1024)
    fmt.Printf("Goroutines: %d → %d\n",
        runtime.NumGoroutine()-10000, runtime.NumGoroutine())
}

// 封装一个带"重置超时"功能的工具
type Timeout struct {
    timer *time.Timer
}

func NewTimeout(d time.Duration) *Timeout {
    return &Timeout{
        timer: time.NewTimer(d),
    }
}

func (t *Timeout) C() <-chan time.Time {
    return t.timer.C
}

func (t *Timeout) Reset(d time.Duration) {
    t.timer.Stop()
    select {
    case <-t.timer.C:
    default:
    }
    t.timer.Reset(d)
}

func (t *Timeout) Stop() {
    t.timer.Stop()
    select {
    case <-t.timer.C:
    default:
    }
}

// 使用
func main() {
    timeout := NewTimeout(5 * time.Second)
    defer timeout.Stop()

    for {
        timeout.Reset(5 * time.Second)

        select {
        case msg := <-ch:
            process(msg)
        case <-timeout.C():
            fmt.Println("超时")
        }
    }
}

flowchart TD
    subgraph 线程T1
        TL1["ThreadLocalMap"]
        TL1 --> E1["Entry
key: 弱引用ThreadLocal
value: 值A"]
    end
    subgraph 线程T2
        TL2["ThreadLocalMap"]
        TL2 --> E2["Entry
key: 弱引用ThreadLocal
value: 值B"]
    end
    subgraph 线程T3
        TL3["ThreadLocalMap"]
        TL3 --> E3["Entry
key: 弱引用ThreadLocal
value: 值C"]
    end

    TH["ThreadLocal对象
（单例）"] -.->|"作为key"| TL1
    TH -.->|"作为key"| TL2
    TH -.->|"作为key"| TL3

// Thread 类中的内部字段
public class Thread implements Runnable {
    ThreadLocal.ThreadLocalMap threadLocals = null;
}

// ThreadLocal 核心方法
public class ThreadLocal<T> {
    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) return (T) e.value;
        }
        return setInitialValue();
    }

    public void set(T value) {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
    }
}

flowchart TD
    A["threadLocal.get()"] --> B["获取当前线程 Thread.currentThread()"]
    B --> C["获取当前线程的 threadLocals (ThreadLocalMap)"]
    C --> D{map 存在?}
    D -->|"存在"| E["以 this (ThreadLocal对象) 为 key 查找 Entry"]
    D -->|"不存在"| F["创建 ThreadLocalMap
调用 initialValue()"]
    E --> G{Entry 存在?}
    G -->|"存在"| H["返回 Entry.value"]
    G -->|"不存在"| I["调用 initialValue()
放入 map"]
    I --> J["返回初始值"]
    H --> K["返回结果"]
    F --> J

static class Entry extends WeakReference<ThreadLocal> {
    Object value;

    Entry(ThreadLocal k, Object v) {
        super(k); // key 为弱引用
        value = v;
    }
}

flowchart TD
    subgraph 正常情况
        REF["ThreadLocal对象
强引用"] --> ACTIVE["可达"]
        MAP_WEAK["Entry.key
弱引用"] -.-> ACTIVE
        ACTIVE --> VAL["value
可访问"]
    end

    subgraph ThreadLocal对象置null后
        NULLED["ThreadLocal对象
置为 null"] --> X[被GC回收]
        MAP_WEAK_KEY["Entry.key
弱引用 → null"] -.-> X
        XN[Entry.value仍存在<br>但无法被访问]
        XN --> MEM["💀 内存泄漏!"]
    end

线程池中的线程复用场景：
1. threadLocal.set(value)  → 向 ThreadLocalMap 存入 Entry
2. 任务执行完毕，但线程被线程池回收复用
3. 没有调用 threadLocal.remove()
4. Entry.key（弱引用）可能已被 GC
5. Entry.value 仍然存活着，线程复用导致 value 无法被回收
6. → 内存泄漏！

/**
 * ThreadLocal 使用示例：存储用户会话信息
 */
public class ThreadLocalDemo {

    // 用户上下文
    private static final ThreadLocal<UserContext> USER_HOLDER = new ThreadLocal<>();

    static class UserContext {
        String userId;
        String userName;

        UserContext(String userId, String userName) {
            this.userId = userId;
            this.userName = userName;
        }
    }

    public static void main(String[] args) {
        // 模拟多个请求，每个请求由不同线程处理
        for (int i = 0; i < 3; i++) {
            final int userId = i;
            new Thread(() -> {
                try {
                    // 设置用户上下文
                    USER_HOLDER.set(new UserContext(
                        "user-" + userId,
                        "用户" + userId
                    ));

                    // 业务处理——从 ThreadLocal 获取上下文
                    UserContext ctx = USER_HOLDER.get();
                    System.out.println(Thread.currentThread().getName() +
                        " 处理请求: " + ctx.userName);

                    // 模拟业务处理
                    Thread.sleep(500);

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // ★ 必须清除！防止内存泄漏
                    USER_HOLDER.remove();
                }
            }, "thread-" + i).start();
        }
    }
}

/**
 * 线程池中 ThreadLocal 的正确使用
 */
public class ThreadLocalInPoolDemo {

    private static final ThreadLocal<Long> COST_TIME = new ThreadLocal<>();

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        for (int i = 0; i < 5; i++) {
            pool.execute(() -> {
                try {
                    // 设置开始时间
                    COST_TIME.set(System.currentTimeMillis());

                    // 模拟业务处理
                    Thread.sleep(200);

                    // 计算耗时
                    long cost = System.currentTimeMillis() - COST_TIME.get();
                    System.out.println(Thread.currentThread().getName() +
                        " 耗时: " + cost + "ms");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // ★ 线程池复用线程，必须清除！
                    COST_TIME.remove();
                }
            });
        }
        pool.shutdown();
    }
}

    com.alibaba
    transmittable-thread-local
    2.14.0

// 使用 TransmittableThreadLocal 实现跨线程池传递
TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();
context.set("parent-value");

// 将 ThreadLocal 包装后提交，值会传递到子线程
Runnable task = () -> System.out.println(context.get());
TtlRunnable ttlTask = TtlRunnable.get(task);
executor.submit(ttlTask);

/**
 * 线程安全的 SimpleDateFormat
 */
public class DateFormatHolder {
    private static final ThreadLocal<SimpleDateFormat> DATE_FORMAT =
        ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    public static String format(Date date) {
        return DATE_FORMAT.get().format(date);
    }

    public static void clear() {
        DATE_FORMAT.remove();
    }
}

public void remove() {
    ThreadLocalMap m = getMap(Thread.currentThread());
    if (m != null)
        m.remove(this); // 删除 Entry，断开 value 引用
}

graph TD
    subgraph "内存泄漏常见来源"
        A["会话上下文膨胀
对话历史不截断"]
        B["工具结果缓存
不设TTL不淘汰"]
        C["异步回调
闭包引用未释放"]
        D["LLM Token缓存
无上限导致OOM"]
        E["连接池泄漏
HTTP/DB连接未归还"]
        F["日志/追踪数据
内存缓冲区过大"]
    end

    A --> Mem["进程内存持续增长"]
    B --> Mem
    C --> Mem
    D --> Mem
    E --> Mem
    F --> Mem

    Mem --> GC["GC压力增大
STW时间变长"]
    GC --> OOM["OOM Kill"]
    Mem --> Swap["Swap频繁
性能雪崩"]

import gc
import tracemalloc
import objgraph  # 需要安装: pip install objgraph
from typing import List, Dict
import weakref

# ====== 问题1：对话上下文无限增长 ======
class LeakyAgent:
    """内存泄漏版本：对话历史不截断"""
    def __init__(self):
        self.conversation_history: List[dict] = []

    async def chat(self, message: str) -> str:
        self.conversation_history.append({"role": "user", "content": message})
        reply = f"Reply to: {message}"
        self.conversation_history.append({"role": "assistant", "content": reply})
        return reply
    # 对话历史会无限膨胀！每天数万条消息全部在内存中

class FixedAgent:
    """修复版：滑动窗口 + 摘要压缩"""
    def __init__(self, max_history: int = 20):
        self.recent_history: List[dict] = []  # 最近的完整对话
        self.summary: str = ""  # 早期对话的摘要
        self.max_history = max_history

    async def chat(self, message: str) -> str:
        self.recent_history.append({"role": "user", "content": message})

        # 超过阈值时压缩
        if len(self.recent_history) > self.max_history:
            # 将最早的一半对话摘要化
            early_msgs = self.recent_history[:self.max_history // 2]
            self.summary = f"{self.summary}\n{self._summarize(early_msgs)}"
            self.recent_history = self.recent_history[self.max_history // 2:]

        reply = f"Reply to: {message}"
        self.recent_history.append({"role": "assistant", "content": reply})
        return reply

    def _summarize(self, messages: List[dict]) -> str:
        return f"[摘要: {len(messages)}条消息]"

# ====== 问题2：循环引用 ======
class ToolResultCache:
    def __init__(self):
        self.cache: dict = {}

    def store_tool_result(self, agent_instance, tool_name: str, result: dict):
        """数据库连接对象被缓存引用，无法GC"""
        # 泄漏点：agent_instance被引用但没有清理机制
        self.cache[f"{id(agent_instance)}:{tool_name}"] = {
            "agent": agent_instance,  # 强引用！
            "result": result,
            "timestamp": time.time()
        }

class FixedToolResultCache:
    def __init__(self, max_entries: int = 1000, ttl: int = 300):
        self.cache: Dict[str, weakref.ref] = {}  # 使用弱引用
        self.max_entries = max_entries
        self.ttl = ttl

    def store_tool_result(self, agent_instance, tool_name: str, result: dict):
        # 使用弱引用避免阻止GC
        key = f"{id(agent_instance)}:{tool_name}"
        self.cache[key] = {
            "agent_ref": weakref.ref(agent_instance),  # 弱引用
            "result": result,
            "timestamp": time.time()
        }
        # 淘汰策略
        if len(self.cache) > self.max_entries:
            oldest = min(self.cache.keys(), 
                        key=lambda k: self.cache[k]["timestamp"])
            del self.cache[oldest]

# ====== 排查工具使用 ======
class MemoryProfiler:
    """内存泄漏排查工具"""

    @staticmethod
    def start_tracking():
        """启动内存追踪"""
        tracemalloc.start(25)  # 保存25层调用栈

    @staticmethod
    def take_snapshot():
        """获取内存快照"""
        return tracemalloc.take_snapshot()

    @staticmethod
    def compare_snapshots(before, after, top_n: int = 10):
        """比较两个快照，找出增长最多的对象"""
        stats = after.compare_to(before, 'lineno')
        print("Top memory allocations growth:")
        for stat in stats[:top_n]:
            print(f"  {stat.count_diff:>8} new blocks, "
                  f"{stat.size_diff / 1024:.1f} KB: {stat.traceback.format()[0]}")

    @staticmethod
    def find_leaked_objects(cls_name: str):
        """查找未被GC回收的特定类实例"""
        # 使用objgraph找出未被回收的对象
        objects = objgraph.by_type(cls_name)
        print(f"Live {cls_name} instances: {len(objects)}")
        if len(objects) > 10:
            # 找出引用链
            objgraph.show_backrefs(objects[:3], max_depth=5)

    @staticmethod
    def check_gc_stats():
        """检查GC统计信息"""
        gc.set_debug(gc.DEBUG_LEAK)
        unreachable = gc.collect()
        print(f"Unreachable objects collected: {unreachable}")
        print(f"GC generations counts: {gc.get_count()}")