Synchronize cuncurrent transactions by criteria in Spring

Question

Need some Guru advice.

Our system checks if the current client's total Debt amount exceeds allowed Credit amount and if true, adds new Debt entry

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

In additionalDebtAllowed() we get all active debt rows by client id and compare with the credit limit, that we get from another system.

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

1. 当前客户债务为50，他的信用额度为100，他又要求50。
2. 两个线程都获得了当前债务（50）。
3. 两个线程都检查信用额度（50 + 50 <= 100）
4. 两个线程都创建新的债务行
5. 现在，客户债务为150，比信用额度还高。

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

考虑了SERIALIZABLE Isolation Level，但是它将锁定整个表，而我仅需要每个客户端进行同步。

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对其进行更多测试，您将看到它分别以正确的方式适用于每个客户端。

希望对您有帮助。