处理spring数据中的并发事务答案

【问题标题】：Handling concurrent transactions in spring data处理spring数据中的并发事务
【发布时间】：2016-08-22 09:22:40
【问题描述】：

我正在使用spring Data。我对spring数据并发事务有如下问题：

实体和仓库如下：

    @Entity
    public class Wallet {

        @Version
        private int version;
        @Id
        @GeneratedValue
        @OrderColumn
        private Long Id;
        @OneToOne()
        @OrderColumn
        private User user;
        @OrderColumn
        private Double virtualBalance;
        @Column(name = "created_by")
        @OrderColumn
        private String createdBy;
        @Column(name = "created_date")
        @OrderColumn
        private Date createdDate;
        @Column(name = "updated_by")
        @OrderColumn
        private String updatedBy;
        @Column(name = "updated_date")
        @OrderColumn
        private Date updatedDate;
... Setters and getters ...
}

repository如下

public interface WalletJpaRepository extends JpaRepository<Wallet, Long>{

    @Lock(LockModeType.OPTIMISTIC) // I have also tried PESSIMISTIC, READ, WRITE, PESSIMISTIC_READ, PESSIMISTIC_WRITE, etc.but they don't seem to work
    Wallet findOne(Long id);

}

我正在同时对其中两个方法进行方法调用，如下所示：

@Test
    public void testConcurrentTransactions() {
        System.out.println("Wallet 1 : ->" + getWallet1());
        System.out.println("Wallet 2 : ->" + getWallet2());
    }

这两种方法如下所述

@Transactional(isolation = Isolation.SERIALIZABLE)
private Wallet getWallet1() {
    Wallet wallet1 = walletJpaRepository.findOne(new Long(1)); // suppose the value of wallet1.getVirtualBalance() is 1000
    wallet1.setVirtualBalance(wallet1.getVirtualBalance().doubleValue() + 100); // After evaluating this line it becomes 1100
    System.out.println(Thread.currentThread().getId());
    return wallet1;
}

@Transactional(isolation = Isolation.SERIALIZABLE)
private Wallet getWallet2() {
    Wallet wallet2 = walletJpaRepository.findOne(new Long(1)); // Here again the value of wallet2.getVirtualBalance() fetched is 1000 but I need 1100 to be the value read
    System.out.println(Thread.currentThread().getId());
    return wallet2;
}

问题是我没有在不同的方法调用中获得同一实体的更新值。

例如，如果 id 为 1 的实体的值在调用 getWallet1() 方法后最初的值为 1000，则该值应更新为 1100，但它不会反映在第二种方法中，即 getWallet2() 并再次反映如上面代码的 cmets 所述，我在第二种方法中得到 1000。

我已尝试使用propagation、Isolation、Lock，但仍然没有得到所需的结果。

有没有办法处理这样的场景，我找不到解决这种情况的方法，这是我在一个巨大的货币交易系统中得到的场景的简化版本，命中率大约是每秒 4 到 5 个事务。

以上只是我尝试重现场景的示例，以下是相同的实际代码。

@Override
@Transactional
public InterWalletRequestFrontendWrapper approveOrDeclineRequest(User requestingUser, String operation,
        String requestId) {

    InterWalletRequest walletRequest = interWalletRequestJpaRepository.findOne(Long.parseLong(requestId));
    if (walletRequest.getStatus().equalsIgnoreCase(Utility.statusInitiated)
            || walletRequest.getStatus().equalsIgnoreCase(Utility.statusPending)) {
        if (operation.equalsIgnoreCase(Utility.operationDecline)) {
            walletRequest.setStatus(Utility.statusDeclined);
            interWalletRequestJpaRepository.save(walletRequest);
            InterWalletRequestFrontendWrapper response = fetchRaisedRequests(requestingUser);
            response.setStatus(0);
            response.setStatusDesc(Utility.statusDeclined);
            return response;
        } else {

            User admin = walletRequest.getRequestTo();
            Wallet adminWallet = admin.getWallet();

            if (adminWallet.getVirtualBalance() >= walletRequest.getAmount()) {
                try {

                    User user = walletRequest.getRequestFrom();

                    UserWalletTransaction txn1 = new UserWalletTransaction();
                    UserWalletTransaction txn2 = new UserWalletTransaction();
                    /**
                     * New transaction initiated for admin
                     */
                    txn1.setAmountTransacted(walletRequest.getAmount());
                    txn1.setDebitUser(admin);
                    txn1.setCreditUser(user);
                    txn1.setOperationPerformed(Utility.operationPerformedInterWallet);
                    txn1.setPreviousAmount(admin.getWallet().getVirtualBalance());
                    txn1.setStatus(Utility.statusNew);
                    txn1.setUser(admin);
                    txn1.setTransactionType(Utility.transactionTypeDebit);
                    txn1.setCreatedBy(admin.getUserName());
                    txn1.setUpdatedBy(admin.getUserName());
                    txn1.setCreatedDate(new Date());
                    txn1.setUpdatedDate(new Date());
                    txn1.setWallet(admin.getWallet());

                    /**
                     * New txn initiated for the user who walletRequested
                     * the txn.
                     */
                    txn2.setAmountTransacted(walletRequest.getAmount());
                    txn2.setDebitUser(admin);
                    txn2.setCreditUser(user);
                    txn2.setOperationPerformed(Utility.operationPerformedInterWallet);
                    txn2.setPreviousAmount(user.getWallet().getVirtualBalance());
                    txn2.setStatus(Utility.statusNew);
                    txn2.setTransactionType(Utility.transactionTypeCredit);
                    txn2.setCreatedBy(admin.getUserName());
                    txn2.setUpdatedBy(admin.getUserName());
                    txn2.setCreatedDate(new Date());
                    txn2.setUpdatedDate(new Date());
                    txn2.setUser(user);
                    txn2.setWallet(user.getWallet());

                    txn2 = walletTransactionJpaRepository.save(txn2);

                    Wallet wallet1 = admin.getWallet();
                    wallet1.setVirtualBalance(admin.getWallet().getVirtualBalance() - walletRequest.getAmount());
                    wallet1 = walletJpaRepository.save(wallet1);

                    /**
                     * After debit set the reference of other user.
                     */

                    txn1.setRelationalTransaction(txn2);
                    /**
                     * After debit from admin set balance amount
                     * 
                     */
                    txn1.setBalanceAmount(wallet1.getVirtualBalance());

                    /**
                     * Money deducted from admin wallet but not credited to
                     * the user wallet. so status is pending.
                     */
                    txn1.setStatus(Utility.statusPending);
                    txn1 = walletTransactionJpaRepository.save(txn1);

                    Wallet wallet2 = user.getWallet();
                    wallet2.setVirtualBalance(user.getWallet().getVirtualBalance() + walletRequest.getAmount());
                    wallet2 = walletJpaRepository.save(wallet2);

                    /**
                     * After credit to User wallet add balance amount.
                     */
                    txn2.setBalanceAmount(wallet2.getVirtualBalance());

                    txn1.setStatus(Utility.statusSuccess);
                    txn2.setStatus(Utility.statusSuccess);
                    txn2.setRelationalTransaction(txn1);

                    List<UserWalletTransaction> transactions = new ArrayList<>();
                    transactions.add(txn1);
                    transactions.add(txn2);

                    walletTransactionJpaRepository.save(transactions);

                    walletRequest.setStatus(Utility.statusApproved);
                    interWalletRequestJpaRepository.save(walletRequest);

                    InterWalletRequestFrontendWrapper response = fetchRaisedRequests(requestingUser);
                    response.setStatus(0);
                    response.setBalance(wallet1.getVirtualBalance());
                    response.setStatusDesc(Utility.statusApproved);
                    return response;

                } catch (Exception e) {
                    System.out.println(".......... Exception Caught ..........");
                    walletRequest.setStatus(Utility.statusPending);
                    interWalletRequestJpaRepository.save(walletRequest);
                    InterWalletRequestFrontendWrapper response = fetchRaisedRequests(requestingUser);
                    response.setStatus(0);
                    response.setStatusDesc(Utility.statusDeclined);
                    return response;
                }
            } else {
                /**
                 * if the admin wallet desn't have enough balance then the
                 * status is set to pending.
                 */
                walletRequest.setStatus(Utility.statusPending);
                interWalletRequestJpaRepository.save(walletRequest);
                InterWalletRequestFrontendWrapper response = fetchRaisedRequests(requestingUser);
                response.setStatus(0);
                response.setStatusDesc(Utility.statusDeclined);
                return response;
            }
        }
    } else {
        InterWalletRequestFrontendWrapper response = fetchRaisedRequests(requestingUser);
        response.setStatus(0);
        response.setStatusDesc(Utility.statusDeclined);
        return response;
    }

}

下面显示了对同一实体进行操作的另一种方法

@Override
@Transactional
private UserWalletTransaction initiateVerifyTransaction(AccountsDetails transfer, User user) {

        Double amountTransacted = 2.00;
        Wallet wallet = user.getWallet();
        UserWalletTransaction transaction = new UserWalletTransaction();
        transaction.setAmountTransacted(amountTransacted);

        transaction.setPreviousAmount(wallet.getVirtualBalance());
        transaction.setOperationPerformed(Utility.operationPerformedDVBeneFundTransfer);
        transaction.setTransactionType(Utility.transactionTypeDebit);

        /**
         * Debit from wallet.
         */
        wallet.setVirtualBalance(wallet.getVirtualBalance() - amountTransacted);
        wallet.setUpdatedDate(new Date());
        wallet.setUpdatedBy(user.getUserName());
        wallet = walletJpaRepository.save(wallet);
        logger.info(wallet);

        transaction.setBalanceAmount(wallet.getVirtualBalance());
        transaction.setUser(user);
        transaction.setWallet(wallet);
        transaction.setStatus(Utility.statusNew);
        transaction.setCreatedBy(user.getUserName());
        transaction.setUpdatedBy(user.getUserName());
        transaction.setCreatedDate(new Date());
         transaction.setToAccount(transfer.getAccount());
         transaction.setBankName(transfer.getBankName());
         transaction.setBeniMobile(transfer.getRecipientMobileNo());
         transaction.setTransactionMode(transfer.getChannel().equalsIgnoreCase("2")
         ? "IMPS" : "NEFT");
        return walletTransactionJpaRepository.save(transaction);

    }

像这样在不同的服务中有七种方法同时访问钱包，因为可以有多个用户同时登录，并且很可能用户管理员也登录并执行货币交易，这就是我们遇到这个问题的真实场景。

提前致谢

【问题讨论】：

首先，您的测试没有并发，也没有事务。您对 Spring AOP 如何工作以及如何应用事务缺乏了解。简而言之，使用代理并且只有方法调用 INTO 一个对象被代理。因此，您从测试用例调用的方法上的@Transactional 基本上是无用的（即使该方法是public）。测试真正的方法，而不是测试用例中的某个东西。接下来，您的服务层应该是事务层，而不是存储库。
感谢 Deinum 指出这些事情。这两种方法实际上是在服务层。但事实是我只是想重现这样的场景以供理解。我的意思是有两个服务使用存储库从数据库同时访问同一个实体。因此，如果要提交最后一个，则在数据库中更新的值。我有兴趣了解如何实现这一点，以便我可以处理并发事务。
您首先提供实际代码...因为getWallet1 不能在服务类中。如果您了解代理，代码将按预期工作。使用乐观锁定时，您依赖于版本号，无需做任何其他事情......
正如我已经提到的，您的服务方法应该是事务性的......从你所拥有的来看，它不是......
你真的读过我的cmets吗...你在private方法上有@Transactional，这不起作用，因为我解释了只有方法调用INTO对象将通过代理。内部方法调用将不起作用。了解代理/aop 的工作原理然后解决您的问题，您缺乏理解。

标签： java spring transactions spring-data spring-data-jpa

【解决方案1】：

大家好，我将回答我自己的问题，这可能会在将来对某人有所帮助，我已经找到了解决问题的方法。感谢 Denium 指出问题。这真是一个很棒的概念。

我犯的错误是对方法进行内部调用并将@Transactional 写入private 方法。

@Transactional 是使用spring AOP 实现的，因此内部方法调用永远不会真正到达代理，@Transactional 的功能行为很奇怪。

所以解决方案是将方法包装在一个对象中并在对象的方法上定义@Transactional，并且只对对象进行外部调用。

其他解决方案可能正在定义我们自己的point cuts 和advice

请随时添加任何建议和修改，

谢谢

【讨论】：