多实例并发访问SQL Server/Azure SQL数据库表时的数据重复问题及事务锁相关疑问

兄弟，你遇到的是典型的并发丢失更新问题——多个实例同时读取同一个计数器，各自加1后写回，结果就撞出重复值了。先给你把核心问题拆明白，再给你靠谱的解决方案：

首先，你原来写的with conn.begin() as transaction:只是开启了一个事务，但默认情况下（SQL Server的READ COMMITTED隔离级别），事务里的读操作不会加排他锁，两个实例还是能同时读到同一个旧值，所以光开事务根本拦不住并发问题。

那该怎么解决？按优先级给你说几个方案：

1. 最优解：用数据库原子操作代替先读后写 #

别在应用层读计数器再加1，直接把增量逻辑丢给数据库执行——SQL Server的UPDATE语句是原子性的，数据库会自动帮你处理并发，不会出现重复。比如你要给某个事件类型的计数器加1，直接写：

from sqlalchemy import text

with engine.connect() as conn:
    with conn.begin():
        # 先尝试更新现有记录
        update_result = conn.execute(
            text("UPDATE Log SET Counter = Counter + 1 WHERE EventType = :event_type; SELECT @@ROWCOUNT;"),
            {"event_type": "你的事件类型"}
        )
        updated_rows = update_result.scalar()
        # 如果没有找到对应记录，就插入新的初始化计数器
        if updated_rows == 0:
            conn.execute(
                text("INSERT INTO Log (EventType, Counter) VALUES (:event_type, 1);"),
                {"event_type": "你的事件类型"}
            )

这种方式完全不用管锁和重试，数据库自己会保证同一时间只有一个实例能成功更新，其他的要么排队要么触发插入，绝对不会有重复。

2. 必须先读后写？那就加显式锁+重试 #

如果业务逻辑必须先读取计数器做一些处理再写回去，那你得在读取的时候就给记录加锁，不让其他实例读。SQL Server里用WITH (UPDLOCK, HOLDLOCK)来实现——UPDLOCK加更新锁，HOLDLOCK把锁持有到事务结束。

同时，被锁阻塞的实例可能会抛出锁超时或者死锁错误，这时候你得在应用层加重试逻辑，比如最多重试3次：

from sqlalchemy import text
import sqlalchemy.exc

def safe_increment_counter(engine, event_type, max_retries=3):
    for attempt in range(max_retries):
        try:
            with engine.connect() as conn:
                with conn.begin():
                    # 读取时加锁，其他实例必须等当前事务提交才能读
                    counter_result = conn.execute(
                        text("SELECT Counter FROM Log WITH (UPDLOCK, HOLDLOCK) WHERE EventType = :event_type;"),
                        {"event_type": event_type}
                    )
                    counter_row = counter_result.fetchone()
                    
                    if counter_row:
                        new_counter = counter_row[0] + 1
                        # 这里可以加你需要的其他业务逻辑
                        conn.execute(
                            text("UPDATE Log SET Counter = :new_counter, OtherData = :other_data WHERE EventType = :event_type;"),
                            {"new_counter": new_counter, "other_data": "你的其他数据", "event_type": event_type}
                        )
                    else:
                        # 初始化新记录
                        conn.execute(
                            text("INSERT INTO Log (EventType, Counter, OtherData) VALUES (:event_type, 1, :other_data);"),
                            {"event_type": event_type, "other_data": "你的其他数据"}
                        )
                # 事务提交成功，直接返回
                return
        except sqlalchemy.exc.OperationalError as e:
            # 捕获SQL Server的死锁（错误码1205）或锁超时错误
            error_msg = str(e)
            if "1205" in error_msg or "lock request time out period exceeded" in error_msg:
                if attempt == max_retries - 1:
                    # 最后一次重试失败，抛异常让上层处理
                    raise
                # 否则继续重试
                continue
        except Exception as e:
            # 其他异常直接抛出
            raise

3. 关于事务隔离级别的补充 #

如果你不想用显式锁，也可以把事务隔离级别改成REPEATABLE READ或者SERIALIZABLE，但SERIALIZABLE隔离级别太严格，容易导致大量锁等待和死锁，除非业务场景特殊，否则不推荐。用SQLAlchemy设置隔离级别可以这么写：

with engine.connect().execution_options(isolation_level="REPEATABLE READ") as conn:
    with conn.begin():
        # 你的业务逻辑
        pass

但还是那句话，优先用原子更新，这是最省心最可靠的方案。

最后提醒下：重试逻辑一定要加，不管用哪种锁机制，并发下都可能出现锁冲突，重试能让失败的请求自动重试，避免业务报错。

备注：内容来源于stack exchange，提问作者BlakeB9