乐观锁与悲观锁的实现和应用

乐观锁与悲观锁：原理、实现与应用详解

在并发编程和数据库操作中，乐观锁和悲观锁是两种重要的并发控制策略，它们在原理、实现方式和应用场景上存在显著差异。下面我们将通过图文结合的方式，深入探讨这两种锁机制。

一、基本概念

1.1 悲观锁

悲观锁的核心思想是先锁后用，它认为在数据处理过程中，很可能会发生并发冲突。因此，在进行数据操作之前，就会获取锁，以确保在当前事务处理期间，其他事务无法对同一数据进行修改，从而保证数据的一致性和完整性。

1.2 乐观锁

乐观锁秉持先试后验的理念，它假定在大多数情况下，数据处理过程中不会发生冲突，所以不会在操作数据前加锁。只有在更新数据时，才会去验证在本次更新之前，是否有其他事务对数据进行了修改。如果没有修改，则执行更新操作；如果数据已被修改，则采取相应的处理措施（如重试、回滚等）。

二、实现方式

2.1 悲观锁的实现

2.1.1 数据库层面

在数据库中，常使用SELECT ... FOR UPDATE语句实现悲观锁。该语句会对查询到的数据加上排它锁（X 锁），阻止其他事务对数据进行读写操作，直到当前事务提交或回滚。

-- 假设存在账户表accounts，包含id, balance字段
CREATE TABLE accounts (
    id INT PRIMARY KEY,
    balance DECIMAL(10, 2) NOT NULL
);
-- 插入测试数据
INSERT INTO accounts (id, balance) VALUES (1, 1000.00);
-- 事务1：扣款操作
BEGIN TRANSACTION;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 假设查询结果: id=1, balance=1000.00
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;
-- 事务2：在事务1提交前尝试扣款
BEGIN TRANSACTION;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 此查询会被阻塞，直到事务1提交或回滚

2.1.2 编程语言层面

在 Java 中，可使用synchronized关键字和ReentrantLock类实现悲观锁；Python 提供了threading.Lock类。这些工具通过互斥访问的方式，保证同一时刻只有一个线程能访问共享资源。

public class PessimisticLockExample {
    private final ReentrantLock lock = new ReentrantLock();
    private double balance = 1000.0;
    public void withdraw(double amount) {
        lock.lock();
        try {
            // 模拟业务处理时间
            Thread.sleep(100);
            if (balance >= amount) {
                balance -= amount;
                System.out.println("扣款成功，余额: " + balance);
            } else {
                System.out.println("余额不足");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        PessimisticLockExample account = new PessimisticLockExample();        // 模拟两个线程同时扣款
        Thread t1 = new Thread(() -> account.withdraw(500));
        Thread t2 = new Thread(() -> account.withdraw(800));        t1.start();
        t2.start();        t1.join();
        t2.join();        System.out.println("最终余额: " + account.balance);
    }
}

2.2 乐观锁的实现

2.2.1 版本号机制

在数据库表中添加一个version字段，每次数据更新时，该字段值递增。更新数据前，先比较当前事务读取的version值与数据库中的version值，若一致则执行更新，并将version值加 1；若不一致，则说明数据已被其他事务修改，本次更新失败。

-- 假设账户表accounts包含id, balance, version字段
CREATE TABLE accounts (
    id INT PRIMARY KEY,
    balance DECIMAL(10, 2) NOT NULL,
    version INT DEFAULT 0
);
-- 插入测试数据
INSERT INTO accounts (id, balance) VALUES (1, 1000.00);
-- 事务1：更新操作
BEGIN TRANSACTION;
SELECT * FROM accounts WHERE id = 1;
-- 返回结果: id=1, balance=1000.00, version=0
UPDATE accounts 
SET balance = balance - 100, version = version + 1 
WHERE id = 1 AND version = 0;
-- 如果执行成功，affected rows = 1，version变为1
COMMIT;
-- 事务2：并发更新操作
BEGIN TRANSACTION;
SELECT * FROM accounts WHERE id = 1;
-- 返回结果: id=1, balance=1000.00, version=0 (因为在事务1提交前读取)
UPDATE accounts 
SET balance = balance - 200, version = version + 1 
WHERE id = 1 AND version = 0;
-- 执行失败，affected rows = 0，因为version已经被事务1更新为1
-- 处理更新失败的逻辑
IF ROW_COUNT() = 0 THEN
    -- 重试或回滚
    ROLLBACK;
END IF;
COMMIT;

2.2.2 时间戳机制

与版本号机制类似，时间戳机制使用数据的最后修改时间来判断数据是否被修改。更新数据时，验证时间戳是否发生变化，若变化则更新失败。

2.2.3 CAS（Compare-and-Swap）操作

CAS 是一种无锁的原子操作，在编程语言和硬件层面均有支持。它包含三个操作数：内存地址（V）、预期原值（A）和新值（B）。仅当内存地址 V 中的值与预期原值 A 相等时，才将内存地址 V 中的值更新为新值 B。

import java.util.concurrent.atomic.AtomicInteger;
public class OptimisticLockExample {
    private AtomicInteger balance = new AtomicInteger(1000);
    public boolean withdraw(double amount) {
        int oldValue;
        int newValue;
        do {
            oldValue = balance.get();
            if (oldValue < amount) {
                System.out.println("余额不足");
                return false;
            }
            newValue = (int) (oldValue - amount);
            // 模拟CAS操作前的竞争
            Thread.yield();
        } while (!balance.compareAndSet(oldValue, newValue));        System.out.println("扣款成功，余额: " + balance.get());
        return true;
    }
    public static void main(String[] args) throws InterruptedException {
        OptimisticLockExample account = new OptimisticLockExample();        // 模拟两个线程同时扣款
        Thread t1 = new Thread(() -> account.withdraw(500));
        Thread t2 = new Thread(() -> account.withdraw(800));        t1.start();
        t2.start();        t1.join();
        t2.join();        System.out.println("最终余额: " + account.balance.get());
    }
}

三、性能对比与应用场景

3.1 性能对比

特性	悲观锁	乐观锁
适用场景	写操作频繁、冲突可能性高的情况	读操作频繁、冲突可能性低的情况
加锁时机	在操作数据之前就加锁	在更新数据的时候才验证
性能表现	会导致较多的锁等待现象，性能开销较大	无需加锁，性能开销较小
实现复杂度	相对简单	相对复杂，需要处理更新失败的情况
典型应用	数据库的行锁、表锁	数据库的版本号、CAS 操作

为了更直观地感受性能差异，可通过以下 Java 代码进行测试：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
public class LockPerformanceTest {
    private static final int THREAD_COUNT = 10;
    private static final int OPS_PER_THREAD = 100000;
    // 悲观锁测试
    static class PessimisticCounter {
        private final ReentrantLock lock = new ReentrantLock();
        private int count = 0;
        public void increment() {
            lock.lock();
            try {
                count++;
            } finally {
                lock.unlock();
            }
        }
        public int getCount() {
            return count;
        }
    }
    // 乐观锁测试
    static class OptimisticCounter {
        private int count = 0;
        public synchronized void increment() {
            count++;
        }
        public int getCount() {
            return count;
        }
    }
    public static void main(String[] args) throws InterruptedException {
        testPessimisticLock();
        testOptimisticLock();
    }
    private static void testPessimisticLock() throws InterruptedException {
        PessimisticCounter counter = new PessimisticCounter();
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        long startTime = System.nanoTime();
        for (int i = 0; i < THREAD_COUNT; i++) {
            executor.submit(() -> {
                for (int j = 0; j < OPS_PER_THREAD; j++) {
                    counter.increment();
                }
            });
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        long endTime = System.nanoTime();
        System.out.println("悲观锁耗时: " + (endTime - startTime) / 1_000_000 + " ms");
        System.out.println("最终计数: " + counter.getCount());
    }
    private static void testOptimisticLock() throws InterruptedException {
        OptimisticCounter counter = new OptimisticCounter();
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        long startTime = System.nanoTime();
        for (int i = 0; i < THREAD_COUNT; i++) {
            executor.submit(() -> {
                for (int j = 0; j < OPS_PER_THREAD; j++) {
                    counter.increment();
                }
            });
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        long endTime = System.nanoTime();
        System.out.println("乐观锁耗时: " + (endTime - startTime) / 1_000_000 + " ms");
        System.out.println("最终计数: " + counter.getCount());
    }
}