悲观行锁

概念定义

两个事务更新同一行不能都赢。悲观策略早解决：先读者持锁，后读者等待。乐观策略晚解决：都执行，提交时一个失败。冲突频繁且重试昂贵时悲观占优；冲突稀少时反之。

工作机制

SELECT FOR UPDATE 取排他行锁，FOR SHARE 让并发读者持共享锁。PostgreSQL 还有 FOR NO KEY UPDATE、NOWAIT 与队列领取专用的 SKIP LOCKED；InnoDB 在高隔离级别下加间隙锁与临键锁，锁的是范围而不只是行。

使用场景

进行中的状态机、带不变量的计数器、先领取再处理的队列。锁定区间要短、全代码路径按规范顺序加锁、设置 lock_timeout 让等待变成可度量的错误。

常见误区

行锁只锁行，不串行化整表。把锁跨外部调用持有，会把争用变成系统性停顿；不同路径以不同顺序加锁，生产流量里就会撞上死锁检测。MVCC 下普通读不阻塞写——很多团队在“读后写”校验从未被保护并出事故后才补 FOR UPDATE。工作队列领取必须配 SKIP LOCKED，否则 worker 在一行上排队。

代码示例

from __future__ import annotations

import sqlite3

SCHEMA = """
CREATE TABLE inventory (
    sku TEXT PRIMARY KEY,
    quantity INTEGER NOT NULL CHECK (quantity >= 0),
    price_cents INTEGER NOT NULL CHECK (price_cents >= 0)
)
"""

UPSERT = """
INSERT INTO inventory (sku, quantity, price_cents)
VALUES (?, ?, ?)
ON CONFLICT(sku) DO UPDATE SET
    quantity = excluded.quantity,
    price_cents = excluded.price_cents
"""


def apply_stock_updates(
    connection: sqlite3.Connection,
    rows: list[tuple[str, int, int]],
) -> int:
    for sku, quantity, price_cents in rows:
        if not sku or quantity < 0 or price_cents < 0:
            raise ValueError(f"invalid inventory row: {sku!r}")

    connection.execute("BEGIN IMMEDIATE")
    try:
        cursor = connection.executemany(UPSERT, rows)
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    return cursor.rowcount


def main() -> None:
    connection = sqlite3.connect(":memory:")
    connection.executescript(SCHEMA)

    first = apply_stock_updates(connection, [("A-1", 5, 1200), ("B-2", 3, 800)])
    second = apply_stock_updates(connection, [("A-1", 9, 1150), ("C-3", 1, 450)])
    print("affected rows:", first, second)
    print(connection.execute(
        "SELECT sku, quantity, price_cents FROM inventory ORDER BY sku"
    ).fetchall())

    try:
        apply_stock_updates(connection, [("D-4", -1, 100)])
    except ValueError as exc:
        print(f"rejected: {exc}")
    assert connection.execute("SELECT COUNT(*) FROM inventory").fetchone()[0] == 3


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)
	}
}