死锁检测与预防

概念定义

锁构成有向图：谁持有、谁在等。图中的环就是死锁——每个成员都在等别人。引擎周期性扫描图，选择回滚代价最小的牺牲者将其回滚，并把错误交还给应用处理。

工作机制

InnoDB 在锁等待开始时查环；PostgreSQL 先等 deadlock_timeout（默认约一秒）再检测，用检测延迟换取锁表繁忙时的 CPU。牺牲者收到特定 SQLSTATE；重试应限次、加抖动、可观测——死锁频率本身就是设计信号。

使用场景

死锁超过阈值说明加锁无序：更新前先排序 id、统一连接顺序、拆分巨型事务。队列领取模式用 SKIP LOCKED 可以完全避开等待边。

常见误区

死锁不是随机损坏而是锁序的确定性环，用交错测试就能复现。不处理顺序、单纯重试牺牲者，负载一高就重新进环。ORM 级联按对象图顺序而非调用顺序加锁，审计必须追真实语句；牺牲者可能落在任一侧，两条路径都要有重试逻辑。持续环状下重试会放大负载，级联成连接池饱和。

代码示例

from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any


class TransactionAborted(RuntimeError):
    pass


@dataclass
class Store:
    data: dict[str, Any] = field(default_factory=dict)
    commits: int = 0
    rollbacks: int = 0

    @contextmanager
    def transaction(self) -> Iterator["Transaction"]:
        snapshot = dict(self.data)
        tx = Transaction(self)
        try:
            yield tx
        except BaseException:
            self.data = snapshot
            self.rollbacks += 1
            raise
        else:
            self.commits += 1


@dataclass
class Transaction:
    store: Store

    def set(self, key: str, value: Any) -> None:
        self.store.data[key] = value

    def delete(self, key: str) -> None:
        self.store.data.pop(key, None)


def main() -> None:
    store = Store(data={"balance": 100})
    with store.transaction() as tx:
        tx.set("balance", 80)
        tx.set("currency", "USD")
    print(store.data, store.commits, store.rollbacks)

    try:
        with store.transaction() as tx:
            tx.set("balance", 0)
            raise TransactionAborted("payment declined")
    except TransactionAborted as error:
        print("rolled back:", error)
    print(store.data, store.commits, store.rollbacks)


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

func seedAndMeasure(ctx context.Context, client *redis.Client) error {
	keys := []string{"metric:hits", "metric:misses", "metric:errors"}

	// 独立命令走管道：一次网络往返
	pipe := client.Pipeline()
	incrCmds := make([]*redis.IntCmd, len(keys))
	for i, key := range keys {
		incrCmds[i] = pipe.Incr(ctx, key)
	}
	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("pipeline exec: %w", err)
	}

	// 事务管道：命令打包进 MULTI/EXEC 原子执行
	tx := client.TxPipeline()
	expireCmds := make([]*redis.BoolCmd, len(keys))
	for i, key := range keys {
		expireCmds[i] = tx.Expire(ctx, key, time.Hour)
	}
	if _, err := tx.Exec(ctx); err != nil {
		return fmt.Errorf("tx exec: %w", err)
	}

	for i, key := range keys {
		fmt.Printf("%s=%d expires=%v\n",
			key, incrCmds[i].Val(), expireCmds[i].Val())
	}
	return nil
}

func main() {
	client := redis.NewClient(&redis.Options{
		Addr:        "127.0.0.1:6379",
		PoolSize:    10,
		ReadTimeout: 500 * time.Millisecond,
	})
	defer client.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	if err := seedAndMeasure(ctx, client); err != nil {
		fmt.Println("failed:", err)
	}
}