乐观锁详解：如何处理高并发下的数据一致性问题

创作时间:

作者:

@小白创作中心

乐观锁详解：如何处理高并发下的数据一致性问题

引用

来源

https://blog.axiaoxin.com/post/optimistic-locking/

在现代应用程序中，特别是在高并发场景下，确保数据一致性是一项重要任务。乐观锁（Optimistic Locking）作为一种有效的并发控制机制，允许多个事务并发地读取相同的数据，而不立即加锁。本文将详细探讨乐观锁的原理、实现方法以及其应用场景，并结合实际示例和相关内容进行讲解。

乐观锁的基本原理

乐观锁的核心思想是“冲突检测”。在数据被修改之前，不对数据加锁。相反，当事务尝试提交时，乐观锁会检测是否有其他事务修改了同一数据。如果检测到冲突，则会回滚事务并提示用户重试操作。这种机制减少了对资源的占用，提高了系统的并发性能。

丢失更新问题

在高并发环境中，丢失更新问题是一个常见的挑战。丢失更新问题发生在两个或多个事务并发地读取和更新相同的数据时。具体来说，如果两个事务读取了相同的数据并分别对其进行更新，后提交的事务可能会覆盖先提交的事务的更改，从而导致数据的部分更新被丢失。

在上图中，假设 Alice 和 Bob 同时尝试从账户中取款。Alice 读取了账户余额为 50 元，并打算取款 40 元。与此同时，Bob 也读取了账户余额为 50 元，并打算取款 30 元。由于没有任何锁机制，Alice 和 Bob 的操作可以并行进行，Alice 认为她可以从账户中提取 40 元，但她没有意识到 Bob 刚刚更改了账户余额，现在账户中只有 20 元，最终可能导致账户余额被错误地更新为 -20 元（50 - 70）。

乐观锁与悲观锁的比较

悲观锁通过在读取数据时立即加锁来防止其他事务修改数据。例如，当 Alice 和 Bob 试图同时读取并更新同一个账户时，悲观锁会阻止 Bob 的更新直到 Alice 提交事务。这种方式减少了冲突的可能性，但会导致较高的锁争用和潜在的死锁问题。

以下是悲观锁处理丢失更新问题的过程图示：

在上图中，Alice 和 Bob 都会对读取的账户表行获取读锁。在 SQL Server 上，使用可重复读（Repeatable Read）或序列化（Serializable）隔离级别时，数据库会获取这些锁。

因为 Alice 和 Bob 都读取了具有主键值为 1 的账户，所以他们中的任何一个在释放读锁之前都不能更改它。这是因为写操作需要获取写锁/排他锁，而读锁/共享锁会阻止获取写锁/排他锁。

只有在 Alice 提交事务并释放账户行上的读锁后，Bob 的 UPDATE 操作才能继续并应用更改。在 Alice 释放读锁之前，Bob 的 UPDATE 操作会被阻塞。

乐观锁则不同，它允许冲突发生，但会在提交更新时检测冲突。为了实现乐观锁，通常会在数据表中增加一个版本号字段。在每次更新时，乐观锁会检查该版本号是否与读取时一致。如果一致，则可以安全地进行更新；如果不一致，则说明数据已被修改，事务会回滚。

乐观锁在实际应用中的处理过程图示如下：

这次，我们有一个额外的版本列。每次执行 UPDATE 或 DELETE 时，版本列都会递增，并且还会在 UPDATE 和 DELETE 语句的 WHERE 子句中使用。为使其工作，我们需要在执行 UPDATE 或 DELETE 之前发出 SELECT 并读取当前版本，否则我们将不知道传递给 WHERE 子句的版本值或递增的值。

应用级事务

如今，通过互联网，我们不再在同一数据库事务的上下文中执行读取和写入，ACID（原子性 Atomicity、一致性 Consistency、隔离性 Isolation、持久性 Durability）也不再足够。例如，考虑以下用例：

没有乐观锁，即使数据库事务使用序列化，也无法捕获这种丢失更新。这是因为读取和写入是在单独的 HTTP 请求中执行的，因此在不同的数据库事务中进行。

因此，乐观锁可以帮助您防止丢失更新，即使在使用包含用户思考时间的应用级事务时也是如此。

实现乐观锁的示例代码

以下是一个使用 Golang 实现乐观锁的示例代码：

package main

import (
    "database/sql"
    "fmt"
    "log"
    "time"
    _ "github.com/go-sql-driver/mysql"
)

type Account struct {
    ID      int
    Balance int
    Version int
}

// updateAccountBalance 尝试更新账户余额，使用乐观锁处理并发冲突
func updateAccountBalance(db *sql.DB, accountID int, amount int) error {
    const maxRetries = 3 // 最大重试次数
    for i := 0; i < maxRetries; i++ {
        tx, err := db.Begin()
        if err != nil {
            return err
        }
        defer tx.Rollback()

        // 获取账户信息
        var account Account
        err = tx.QueryRow("SELECT id, balance, version FROM accounts WHERE id = ?", accountID).Scan(&account.ID, &account.Balance, &account.Version)
        if err != nil {
            tx.Rollback()
            return err
        }

        // 更新余额并增加版本号
        newBalance := account.Balance + amount
        newVersion := account.Version + 1
        result, err := tx.Exec("UPDATE accounts SET balance = ?, version = ? WHERE id = ? AND version = ?", newBalance, newVersion, account.ID, account.Version)
        if err != nil {
            tx.Rollback()
            return err
        }

        // 检查更新是否成功
        rowsAffected, err := result.RowsAffected()
        if err != nil {
            tx.Rollback()
            return err
        }
        if rowsAffected == 0 {
            tx.Rollback()
            // 如果更新失败，说明乐观锁冲突，等待一段时间后重试
            time.Sleep(time.Second)
            continue
        }

        // 提交事务
        err = tx.Commit()
        if err != nil {
            return err
        }
        return nil
    }
    return fmt.Errorf("重试次数达到上限，更新账户余额失败")
}

func main() {
    dsn := "user:password@tcp(127.0.0.1:3306)/testdb"
    db, err := sql.Open("mysql", dsn)
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    err = updateAccountBalance(db, 1, 50)
    if err != nil {
        log.Println("更新账户余额失败：", err)
    } else {
        log.Println("账户余额更新成功")
    }
}