发布于 由admin

JAVA高并发下频繁超卖问题：锁粒度优化与乐观锁冲突排查

JAVA高并发下频繁超卖问题：锁粒度优化与乐观锁冲突排查

大家好，今天我们来聊聊在高并发环境下，Java应用程序中常见的超卖问题，并深入探讨如何通过锁粒度优化和乐观锁冲突排查来解决它。超卖问题是指在库存有限的情况下，由于并发操作导致销售数量超过实际库存，造成经济损失。

超卖问题场景分析

超卖问题在高并发的电商、票务等系统中非常普遍。我们来看一个简单的秒杀场景：

假设一个商品库存为10，多个用户同时发起购买请求。如果处理不当，可能出现卖出11件甚至更多的情况。

以下是一个简单的模拟超卖问题的代码：

public class Inventory {

    private int stock = 10;

    public boolean decreaseStock(int quantity) {
        if (stock >= quantity) {
            stock -= quantity;
            System.out.println("成功购买 " + quantity + " 件，剩余库存：" + stock);
            return true;
        } else {
            System.out.println("库存不足，购买失败！剩余库存：" + stock);
            return false;
        }
    }

    public int getStock() {
        return stock;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }).start();
        }
    }
}

这段代码在单线程环境下没问题，但在多线程环境下，由于stock >= quantity判断和stock -= quantity操作不是原子性的，可能出现以下情况：

  1. 线程A读取stock值为1。
  2. 线程B读取stock值也为1。（此时两个线程都认为可以购买）
  3. 线程A执行stock -= quantity，stock变为0。
  4. 线程B执行stock -= quantity，stock变为-1。（超卖发生）

锁粒度优化：从粗到细

解决超卖问题最直接的方法就是加锁，保证操作的原子性。但锁的粒度直接影响并发性能。

1. 粗粒度锁（悲观锁）：synchronized 或 ReentrantLock

最简单的做法是对整个decreaseStock方法加锁：

public class Inventory {

    private int stock = 10;
    private final Object lock = new Object();

    public boolean decreaseStock(int quantity) {
        synchronized (lock) {
            if (stock >= quantity) {
                stock -= quantity;
                System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + stock);
                return true;
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock);
                return false;
            }
        }
    }

    public int getStock() {
        return stock;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

或者使用ReentrantLock:

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Inventory {

    private int stock = 10;
    private final Lock lock = new ReentrantLock();

    public boolean decreaseStock(int quantity) {
        lock.lock();
        try {
            if (stock >= quantity) {
                stock -= quantity;
                System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + stock);
                return true;
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock);
                return false;
            }
        } finally {
            lock.unlock();
        }
    }

    public int getStock() {
        return stock;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

这种方式简单粗暴，保证了线程安全，解决了超卖问题。但所有线程都要竞争同一把锁，并发性能很差。所有购买请求都必须串行执行，在高并发场景下会成为瓶颈。

2. 细粒度锁：只锁 critical section

我们可以尝试缩小锁的范围，只锁住修改stock变量的部分：

public class Inventory {

    private int stock = 10;
    private final Object lock = new Object();

    public boolean decreaseStock(int quantity) {
        if (stock >= quantity) { // 先判断，避免不必要的锁竞争
            synchronized (lock) {
                if (stock >= quantity) { // 双重检查，防止其他线程已经修改了stock
                    stock -= quantity;
                    System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + stock);
                    return true;
                } else {
                    System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock);
                    return false;
                }
            }
        } else {
            System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock);
            return false;
        }
    }

    public int getStock() {
        return stock;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

这里使用了双重检查锁（Double-Checked Locking，DCL），先进行一次stock >= quantity的判断，只有在可能需要修改stock时才进入synchronized块。这样可以减少锁的竞争，提高并发性能。但是需要注意，DCL 在某些 JVM 版本中可能存在问题，需要将 stock 声明为 volatile 来避免指令重排序问题。

3. 更细粒度的锁：使用 StripedLock

如果商品种类很多，不同的商品之间购买请求互不影响，可以考虑使用 StripedLock (也叫分段锁)。 StripedLock 将一把锁拆分成多个锁，每个锁保护不同的商品。这样，不同商品的购买请求可以并发执行，进一步提高并发性能。

虽然Java标准库没有直接提供 StripedLock 的实现，但我们可以使用 Guava 库中的 Striped 类来实现：

import com.google.common.util.concurrent.Striped;

import java.util.concurrent.locks.Lock;

public class Inventory {

    private int stock = 10;
    private final Striped<Lock> stripedLock = Striped.lock(16); // 16个锁，可以根据实际情况调整

    public boolean decreaseStock(int quantity) {
        Lock lock = stripedLock.get(this); // 根据商品ID获取对应的锁
        lock.lock();
        try {
            if (stock >= quantity) {
                stock -= quantity;
                System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + stock);
                return true;
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock);
                return false;
            }
        } finally {
            lock.unlock();
        }
    }

    public int getStock() {
        return stock;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

在这个例子中，我们使用 Striped.lock(16) 创建了 16 个锁。stripedLock.get(this) 根据 Inventory 对象的 hashCode 选择一个锁。这样，不同的 Inventory 对象（代表不同的商品）就可以使用不同的锁，提高并发性能。

锁粒度对比：

锁粒度 优点 缺点 适用场景
粗粒度锁 实现简单，易于理解和维护 并发性能差，所有线程竞争同一把锁 并发量较低，对性能要求不高的场景
细粒度锁 并发性能有所提升，减少了锁的竞争 实现相对复杂，需要考虑线程安全问题，如DCL需要volatile修饰变量，容易出错 并发量较高，对性能有一定要求的场景
更细粒度锁 并发性能大幅提升，不同商品可以并发购买 实现复杂，需要引入额外的库（如Guava），需要根据实际情况选择合适的锁数量，hash冲突严重时性能会下降 商品种类很多，不同商品之间的购买请求互不影响，对并发性能要求极高的场景

乐观锁：无锁编程与冲突检测

除了悲观锁，我们还可以使用乐观锁来解决超卖问题。乐观锁认为在大部分情况下不会发生冲突，因此不会主动加锁，而是在更新数据时检查是否被其他线程修改过。

1. 版本号机制

最常见的乐观锁实现方式是使用版本号（Version）。在数据库表中增加一个版本号字段，每次更新数据时，版本号加1。在更新数据时，需要比较当前版本号和数据库中的版本号是否一致，如果一致则更新成功，否则更新失败。

public class Inventory {

    private int stock = 10;
    private int version = 0;

    public synchronized boolean decreaseStock(int quantity) {
            if (stock >= quantity) {
                stock -= quantity;
                version++;
                System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + stock + ", version: " + version);
                return true;
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + stock + ", version: " + version);
                return false;
            }
    }

    public int getStock() {
        return stock;
    }

    public int getVersion() {
        return version;
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

但这个代码只是模拟概念，真正的乐观锁版本号机制需要结合数据库操作。以下是一个结合数据库的示例：

public class Inventory {

    private int id;
    private int stock;
    private int version;

    // 构造函数，Getter和Setter省略

    public boolean decreaseStock(int quantity) {
        // 1. 从数据库中读取库存和版本号
        Inventory inventory = inventoryDao.getInventoryById(id);
        int currentStock = inventory.getStock();
        int currentVersion = inventory.getVersion();

        // 2. 判断库存是否足够
        if (currentStock >= quantity) {
            // 3. 计算新的库存和版本号
            int newStock = currentStock - quantity;
            int newVersion = currentVersion + 1;

            // 4. 更新数据库
            int rows = inventoryDao.updateStock(id, newStock, newVersion, currentVersion);

            // 5. 判断更新是否成功
            if (rows > 0) {
                System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + newStock + ", version: " + newVersion);
                return true;
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存更新失败，可能已被其他线程修改");
                return false;
            }
        } else {
            System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + currentStock + ", version: " + currentVersion);
            return false;
        }
    }
}

// DAO 接口
interface InventoryDao {
    Inventory getInventoryById(int id);
    int updateStock(int id, int stock, int newVersion, int currentVersion);
}

// DAO 实现类 (示例)
class InventoryDaoImpl implements InventoryDao {
    // 模拟数据库操作
    @Override
    public Inventory getInventoryById(int id) {
        // 假设从数据库中查询到库存为 10，版本号为 0
        Inventory inventory = new Inventory();
        inventory.setId(id);
        inventory.setStock(10);
        inventory.setVersion(0);
        return inventory;
    }

    @Override
    public int updateStock(int id, int stock, int newVersion, int currentVersion) {
        // 模拟更新数据库
        // 假设数据库中存在一条记录，id=1，stock=10，version=0
        // 使用 SQL 语句：UPDATE inventory SET stock = ?, version = ? WHERE id = ? AND version = ?
        if (currentVersion == 0) {
            // 模拟更新成功
            return 1;
        } else {
            // 模拟更新失败，因为版本号不一致
            return 0;
        }
    }
}

// 模拟测试
public class Main {
    public static void main(String[] args) {
        Inventory inventory = new Inventory();
        inventory.setId(1);
        InventoryDao inventoryDao = new InventoryDaoImpl();

        // 将 InventoryDao 注入到 Inventory 类中
        // 在实际应用中，可以使用 Spring 等 IoC 容器来管理 Bean
        inventory.inventoryDao = inventoryDao;

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

对应的数据库 SQL 语句类似：

UPDATE inventory SET stock = newStock, version = newVersion WHERE id = itemId AND version = oldVersion;

如果UPDATE语句返回的受影响行数为0，则表示更新失败，需要重试。

2. CAS（Compare and Swap）

CAS 是一种无锁算法，它通过比较内存中的值和预期值是否相等，如果相等则更新为新值。Java中的java.util.concurrent.atomic包提供了CAS相关的类，如AtomicInteger、AtomicLong等。

import java.util.concurrent.atomic.AtomicInteger;

public class Inventory {

    private AtomicInteger stock = new AtomicInteger(10);

    public boolean decreaseStock(int quantity) {
        int currentStock = stock.get();
        while (true) {
            if (currentStock >= quantity) {
                int newStock = currentStock - quantity;
                if (stock.compareAndSet(currentStock, newStock)) {
                    System.out.println(Thread.currentThread().getName() + " 成功购买 " + quantity + " 件，剩余库存：" + newStock);
                    return true;
                } else {
                    // CAS 失败，说明有其他线程修改了 stock，重新获取最新值并重试
                    currentStock = stock.get();
                }
            } else {
                System.out.println(Thread.currentThread().getName() + " 库存不足，购买失败！剩余库存：" + currentStock);
                return false;
            }
        }
    }

    public int getStock() {
        return stock.get();
    }

    public static void main(String[] args) {
        Inventory inventory = new Inventory();

        // 模拟并发购买
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                inventory.decreaseStock(1);
            }, "Thread-" + i).start();
        }
    }
}

在这个例子中，我们使用 AtomicInteger 来保证 stock 变量的原子性操作。compareAndSet 方法会比较当前值和预期值是否相等，如果相等则更新为新值，并返回 true；否则返回 false。如果 CAS 失败，说明有其他线程修改了 stock，我们需要重新获取最新值并重试。

乐观锁的优点：

  • 无锁，避免了锁的竞争，提高了并发性能。
  • 适用于读多写少的场景，冲突概率较低。

乐观锁的缺点：

  • 需要重试机制，如果冲突概率很高，会导致大量的重试，反而降低性能。
  • ABA 问题：如果一个变量的值从 A 变成 B，又从 B 变成 A，CAS 认为变量没有被修改过，但实际上可能已经发生了变化。可以使用 AtomicStampedReference 来解决 ABA 问题。

乐观锁的选择：

  • 如果并发量不高，冲突概率较低，可以使用版本号机制。
  • 如果需要原子性操作，可以使用 CAS。

乐观锁冲突排查

在高并发环境下，即使使用了乐观锁，仍然可能出现冲突，导致更新失败。我们需要排查冲突的原因，并采取相应的措施。

1. 日志分析

在代码中添加详细的日志，记录每次更新操作的参数和结果，以及冲突发生的时间和原因。通过分析日志，可以了解冲突发生的频率和模式，从而找到问题的根源。

2. 监控指标

监控更新操作的成功率和重试次数。如果成功率很低，或者重试次数很高，说明冲突很严重，需要优化代码或者调整参数。

3. 模拟并发

使用工具模拟高并发场景，重现冲突问题。可以使用 JMeter、LoadRunner 等工具来模拟大量的并发请求，从而测试代码的并发性能和稳定性。

4. 代码审查

仔细审查代码，查找可能导致冲突的原因。例如，是否存在多个线程同时修改同一条数据，或者是否存在长时间占用资源的操作。

5. 数据库分析

如果使用了数据库，可以分析数据库的锁等待情况和事务执行情况，了解是否存在锁冲突或者死锁。

常见的冲突原因和解决方法：

冲突原因 解决方法
并发量过高 1. 优化代码，减少锁的竞争。 2. 增加服务器数量，提高系统的整体处理能力。 3. 使用缓存，减少数据库的访问压力。
锁粒度过粗 1. 缩小锁的范围，只锁住 critical section。 2. 使用更细粒度的锁，如 StripedLock。
长时间占用资源 1. 优化代码，减少单个请求的处理时间。 2. 使用异步处理，将耗时操作放到后台执行。
多个线程同时修改同一条数据 1. 重新设计数据结构，避免多个线程同时修改同一条数据。 2. 使用消息队列，将更新操作串行化。
数据库锁冲突或死锁 1. 优化 SQL 语句，减少锁的持有时间。 2. 调整数据库参数，如锁超时时间。 3. 避免长事务，尽量将事务拆分成多个小事务。 4. 使用乐观锁，减少数据库的锁竞争。
业务逻辑复杂，重试次数过多 1. 优化业务逻辑，减少冲突发生的概率。 2. 增加重试次数的上限，避免无限重试。 3. 引入熔断机制，当重试次数超过上限时，直接返回失败，避免系统崩溃。

选择合适的锁策略

选择合适的锁策略需要根据实际情况进行权衡。没有一种锁策略是万能的，只有最适合当前场景的锁策略。

以下是一些选择锁策略的建议：

  • 并发量低，对性能要求不高： 可以使用粗粒度锁，如 synchronized 或 ReentrantLock。
  • 并发量较高，对性能有一定要求： 可以使用细粒度锁，如双重检查锁。
  • 商品种类很多，不同商品之间的购买请求互不影响，对并发性能要求极高： 可以使用更细粒度的锁，如 StripedLock。
  • 读多写少，冲突概率较低： 可以使用乐观锁，如版本号机制或 CAS。
  • 写多读少，冲突概率较高： 建议使用悲观锁，避免大量的重试。

同时，需要根据实际情况进行性能测试，选择最优的锁策略。

总结：优化锁粒度，灵活选择锁策略

在高并发环境下，超卖问题是一个常见的挑战。通过优化锁粒度，可以有效地提高并发性能，避免超卖问题的发生。同时，需要根据实际情况选择合适的锁策略，并对冲突进行排查和解决，才能保证系统的稳定性和可靠性。

希望今天的分享对大家有所帮助，谢谢！

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注