在 Spring 中按条件同步并发事务

Question

需要一些大师的建议。

我们的系统检查当前客户端的total Debt amount 是否超过允许的Credit amount，如果为真，则添加新的Debt 条目

if (additionalDebtAllowed(clientId, amount)) {
    deptRepository.saveAndFlush(new Debt(clientId, amount));
}

在 additionalDebtAllowed() 中，我们按客户 ID 获取所有活跃的债务行，并与我们从另一个系统获得的信用额度进行比较。

问题是 REST 调用可能是并发的，我们可以在以下情况下运行：

1. 当前客户债务为 50，他的信用额度为 100，他要求 另外 50 个。
2. 两个线程都获得当前债务 (50)。
3. 两个线程都检查信用额度 (50 + 50
4. 两个线程都会创建新的债务行
5. 现在客户债务为 150，超过信用额度。

最简单的方法是在读取和持久化数据之前尝试通过客户端 ID 锁定数据库中的某些行。如果成功 - 继续并解锁。如果失败 - 重试直到成功。但我认为可能还有更漂亮的方式。

考虑了 SERIALIZABLE 隔离级别，但它会锁定整个表，而我只需要每个客户端同步。

Answer 1

我会尽量用简单的方式来做，而不是把事情复杂化。

我将专注于真正的问题，而不是代码的美。

我已经测试过的方法如下：

我创建了一个主类，其中两个 CompletableFuture 模拟了对同一个 clientId 的两个同时调用。

//Simulate lines of db debts per user
static List<Debt> debts = new ArrayList<>();

static Map<String, Object> locks = new HashMap<String, Object>();

public static void main(String[] args) {

    String clientId = "1";

    //Simulate previous insert line in db per clientId
    debts.add(new Debt(clientId,50));

    //In a operation, put in a map the clientId to lock this id
    locks.put(clientId, new Object());

    final ExecutorService executorService = Executors.newFixedThreadPool(10);

    CompletableFuture.runAsync(() -> {
        try {
            operation(clientId, 50);
        } catch (Exception e) {
        }
    }, executorService);

    CompletableFuture.runAsync(() -> {
        try {
            operation(clientId, 50);
        } catch (Exception e) {
        }
    }, executorService);

    executorService.shutdown();
}

方法操作是关键。我已经通过clientId同步了地图，这意味着对于其他clientId它不会被锁定，对于每个clientId它都会同时传递一个线程。

private static void operation(String clientId, Integer amount) {
    System.out.println("Entra en operacion");
    synchronized(locks.get(clientId)) {
        if(additionalDebtAllowed(clientId, 50)) {
            insertDebt(clientId, 50);
        }
    }
}

以下方法模拟插入、数据库搜索和远程搜索，但我认为这个概念已经理解，我可以使用存储库来实现，但这不是重点。

private static boolean additionalDebtAllowed(String clientId, Integer amount) {

    List<Debt> debts = debtsPerClient(clientId);

    int sumDebts = debts.stream().mapToInt(d -> d.getAmount()).sum();

    int limit = limitDebtPerClient(clientId);

    if(sumDebts + amount <= limit) {
        System.out.println("Debt accepted");
        return true;
    }

    System.out.println("Debt denied");

    return false;
}

//Simulate insert in db
private static void insertDebt(String clientId, Integer amount) {
    debts.add(new Debt(clientId, amount));
}

//Simulate search in db
private static List<Debt> debtsPerClient(String clientId) {

    return debts;
}

//Simulate rest petition limit debt
private static Integer limitDebtPerClient(String clientId) {

    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    return 100;
}

您可以使用另一个 clientId 和另一个 CompletableFuture 进行更多测试，您会发现它以正确的方式分别适用于每个客户端。

希望对你有帮助。