Uber 如何用 250ms 批量窗口让账本系统扛住每秒 30+ 次热账户更新

1. 读取账户当前余额 balance_old（带版本号 v）
2. 计算新余额 balance_new = balance_old + net_delta
3. 写入：UPDATE account SET balance = balance_new, version = v+1
        WHERE account_id = X AND version = v
4. 如果 WHERE 条件不匹配（版本已被改），重试步骤 1-3

pip install redis
redis-server  # 默认 6379 端口

import redis
import threading
import time
import uuid
from dataclasses import dataclass

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

BATCH_WINDOW_MS = 250
ACCOUNT_ID = "user_42"

# ── 1. 写入端：把变更追加到 Redis 缓冲区 ──
def submit_delta(account_id: str, delta: float, txn_id: str):
    """模拟一笔交易到来，把金额变更 INCR 到 Redis 缓冲区"""
    epoch = int(time.time() * 1000 // BATCH_WINDOW_MS)
    buffer_key = f"ledger:buffer:{account_id}:{epoch}"
    # 用 INCRBYFLOAT 原子追加差额（正数=入账，负数=扣款）
    r.incrbyfloat(buffer_key, delta)
    # 记录这笔交易属于哪个 epoch，用于审计追溯
    r.sadd(f"ledger:txns:{account_id}:{epoch}", txn_id)
    # 设置缓冲区 key 的 TTL，防止孤儿数据
    r.pexpire(buffer_key, BATCH_WINDOW_MS * 3)
    r.pexpire(f"ledger:txns:{account_id}:{epoch}", BATCH_WINDOW_MS * 3)
    print(f"  → txn {txn_id}: delta={delta}, epoch={epoch}")


# ── 2. 批处理器：每 250ms 收割缓冲区，乐观原子写入 ──
def batch_processor(account_id: str, stop_event: threading.Event):
    """每 250ms 收割当前 epoch 的缓冲区，合并写入持久层"""
    while not stop_event.is_set():
        time.sleep(BATCH_WINDOW_MS / 1000.0)
        epoch = int(time.time() * 1000 // BATCH_WINDOW_MS)
        buffer_key = f"ledger:buffer:{account_id}:{epoch}"
        txn_set_key = f"ledger:txns:{account_id}:{epoch}"

        net_delta = r.get(buffer_key)
        if net_delta is None:
            continue  # 本窗口无变更，跳过

        net_delta = float(net_delta)
        txn_ids = r.smembers(txn_set_key)

        # 乐观原子更新（模拟 CAS）
        max_retries = 5
        for attempt in range(max_retries):
            # 读取当前余额和版本
            current = r.hgetall(f"ledger:account:{account_id}")
            if not current:
                current = {"balance": "0.0", "version": "0"}
            old_balance = float(current["balance"])
            old_version = int(current["version"])

            new_balance = old_balance + net_delta
            new_version = old_version + 1

            # CAS 写入：用 Lua 脚本保证原子性
            lua_cas = """
            local key = KEYS[1]
            local expected_version = tonumber(ARGV[1])
            local new_balance = ARGV[2]
            local new_version = tonumber(ARGV[3])

            local current_version = tonumber(redis.call('HGET', key, 'version') or '0')
            if current_version == expected_version then
                redis.call('HSET', key, 'balance', new_balance, 'version', new_version)
                return 1
            else
                return 0
            end
            """
            success = r.eval(lua_cas, 1,
                             f"ledger:account:{account_id}",
                             old_version, str(new_balance), new_version)

            if success:
                # 写入成功，清理缓冲区
                r.delete(buffer_key, txn_set_key)
                # 写入审计日志
                audit_entry = f"epoch={epoch}|delta={net_delta}|txns={','.join(txn_ids)}|new_balance={new_balance}"
                r.rpush(f"ledger:audit:{account_id}", audit_entry)
                print(f"  ✓ BATCH COMMIT: epoch={epoch}, net_delta={net_delta}, "
                      f"txns={len(txn_ids)}, balance={old_balance}→{new_balance}")
                break
            else:
                print(f"  ⚠ CAS conflict (attempt {attempt+1}), retrying...")
                time.sleep(0.01)  # 短暂等待后重试


# ── 3. 模拟并发写入 ──
def simulate_concurrent_writes(account_id: str, num_txns: int):
    """模拟多线程并发提交交易"""
    threads = []
    for i in range(num_txns):
        delta = 10.0 if i % 3 == 0 else -5.0  # 模拟混合入账/扣款
        txn_id = str(uuid.uuid4())[:8]
        t = threading.Thread(target=submit_delta,
                             args=(account_id, delta, txn_id))
        threads.append(t)
        t.start()
        time.sleep(0.03)  # 约 30ms 一笔，模拟 30+ updates/s
    for t in threads:
        t.join()


# ── 4. 运行演示 ──
if __name__ == "__main__":
    # 清理测试数据
    for key in r.keys(f"ledger:*:{ACCOUNT_ID}*"):
        r.delete(key)

    # 启动批处理器
    stop = threading.Event()
    processor = threading.Thread(target=batch_processor,
                                 args=(ACCOUNT_ID, stop))
    processor.start()

    # 模拟 1 秒内约 30 笔并发写入
    print("=== Simulating ~30 concurrent updates in 1s ===")
    simulate_concurrent_writes(ACCOUNT_ID, 30)

    # 等待最后一个窗口处理完
    time.sleep(0.5)
    stop.set()
    processor.join()

    # 打印最终状态
    final = r.hgetall(f"ledger:account:{ACCOUNT_ID}")
    audit_log = r.lrange(f"ledger:audit:{ACCOUNT_ID}", 0, -1)
    print(f"\n=== Final State ===")
    print(f"Balance: {final.get('balance', '0.0')}, Version: {final.get('version', '0')}")
    print(f"Audit entries: {len(audit_log)}")
    for entry in audit_log:
        print(f"  {entry}")