Параллельная обработка: синхронизация против нескольких вызовов БД - PullRequest
Новая
       111

Параллельная обработка: синхронизация против нескольких вызовов БД

1 голос
/ 27 января 2020

Представьте себе банк MN C, который хочет внедрить API перевода учетной записи, используя только ядро ​​Java, и API будет использоваться в многопоточной среде, и поддерживать постоянство суммы счета все время без каких-либо тупиков, конечно

Я имею в виду два подхода для реализации этого и знаю их плюсы и минусы, как описано ниже, но не могу измерить, какие плюсы здесь более хороши в случае параллелизма или когда у нас больше нагрузки. Или, какие минусы должны повлиять больше в случае параллелизма или когда у нас больше нагрузки.

Пожалуйста, дайте совет и предоставьте ваше предложение.

Полный код ссылки на репозиторий Github https://github.com/sharmama07/money-transfer

Прототип API: public boolean transferAmount(Integer fromAccountId, Integer toAccountId, Integer amount);

Подход 1. Обработка параллелизма с помощью операторов while l oop и SQL для проверки предыдущего баланса в предложении where. Если предыдущий баланс не совпадает, вызов суммы обновления для учетной записи не удастся, и он получит сумму последней учетной записи из БД и попытается обновить ее, пока не обновит ее успешно. Здесь ни один поток не будет заблокирован, что означает отсутствие шансов взаимоблокировки, никаких издержек приостановки потока и никакой задержки потока, но у него может быть немного больше вызовов БД 

public boolean transferAmount(Integer fromAccountId, Integer toAccountId, Double amount) {

        boolean updated = false;

        try {
            while(!updated) {

                Account fromAccount = accountDAO.getAccount(fromAccountId);

                if(fromAccount.getAmount()-amount < 0) {throw new OperationCannotBePerformedException("Insufficient balance!");}

                int recordsUpdated = accountDAO.updateAccountAmount(fromAccount.getId(), fromAccount.getAmount(), 
                        fromAccount.getAmount()-amount);
                updated = (recordsUpdated==1);

            }
        }catch (OperationCannotBePerformedException e) {
            LOG.log(Level.SEVERE, "Debit Operation cannot be performed, because " + e.getMessage());
        }

        if(updated) {
            updated = false;
            try {
                while(!updated) {
                    Account toAccount = accountDAO.getAccount(toAccountId);
                    int recordsUpdated = accountDAO.updateAccountAmount(toAccount.getId(), toAccount.getAmount(), toAccount.getAmount()+amount);
                    updated = (recordsUpdated==1);
                }
            }catch (OperationCannotBePerformedException e) {
                LOG.log(Level.SEVERE, "Credit Operation cannot be performed, because " + e.getMessage());
                revertDebittransaction(fromAccountId, amount);
            }
        }


        return updated;
    }

// Account DAO call
@Override
    public Account getAccount(Integer accountId) throws OperationCannotBePerformedException {
        String SQL = "select id, amount from ACCOUNT where id="+accountId+"";
        ResultSet rs;
        try {
            rs = statement.executeQuery(SQL);
            if (rs.next()) {
                return new Account(rs.getInt(1), rs.getDouble(2));
            }
            return null;
        } catch (SQLException e) {
            LOG.error("Cannot retrieve account from DB, reason: "+ e.getMessage());
            throw new OperationCannotBePerformedException("Cannot retrieve account from DB, reason: "+ e.getMessage(), e);
        }

    }

    @Override
    public int updateAccountAmount(Integer accountId, Double currentAmount, Double newAmount) throws OperationCannotBePerformedException {
        String SQL = "update ACCOUNT set amount=" + newAmount +" where id="+accountId+" and amount="+currentAmount+"";
        int rs;
        try {
            rs = statement.executeUpdate(SQL);
            return rs;
        } catch (SQLException e) {
            LOG.error("Cannot update account amount, reason: "+ e.getMessage());
            throw new OperationCannotBePerformedException("Cannot update account amount, reason: "+ e.getMessage(), e);
        }
    }

Подход 2: Здесь другие потоки будут заблокированы, если одна и та же учетная запись находится в двух транзакциях в другом потоке,
Но при этом будет меньше вызовов БД 

    public boolean transferAmount1(Integer fromAccountId, Integer toAccountId, Double amount) {

            boolean updated = false;
            Integer smallerAccountId = (fromAccountId<toAccountId)? fromAccountId: toAccountId;
            Integer largerAccountId =  (fromAccountId<toAccountId)? toAccountId:fromAccountId;

            synchronized(smallerAccountId) {
                synchronized(largerAccountId) {
                    try {
                        Account fromAccount = accountDAO.getAccount(fromAccountId);
                        if(fromAccount.getAmount()-amount < 0) {
                            throw new OperationCannotBePerformedException("Insufficient balance!");
                        }
                        int recordsUpdated = accountDAO.updateAccountAmount(fromAccount.getId(),
                            fromAccount.getAmount(), fromAccount.getAmount()-amount);
                        updated = (recordsUpdated==1);
                    }catch (OperationCannotBePerformedException e) {
                        LOG.log(Level.SEVERE, "Debit Operation cannot be performed, because " + e.getMessage());
                    }
                    // credit operation
                    if(updated) {
                        try {
                            updated = false;
                            Account toAccount = accountDAO.getAccount(toAccountId);
                            int recordsUpdated = accountDAO.updateAccountAmount(toAccount.getId(),
                                toAccount.getAmount(), toAccount.getAmount()+amount);
                            updated = (recordsUpdated==1);
                        }catch (OperationCannotBePerformedException e) {
                            LOG.log(Level.SEVERE, "Credit Operation cannot be performed, because " + e.getMessage());
                            revertDebittransaction(fromAccountId, amount);
                        }
                    }
                }
            }

            return updated;
    }

1 Ответ

0 голосов
/ 03 февраля 2020

Полагаю, мы по какой-то причине предполагаем, что транзакции с БД не подходят для этой схемы. Потому что с транзакциями БД ничего из этого не требуется.

Первая схема должна работать, но она небезопасна. Если первое обновление l oop работает, а второе выходит из строя, первый счет теряет деньги, не переводя его на второй. Хотя есть и лучшие способы справиться с этим. Циклы чтения-операции-обновления по своей природе являются редкими, поэтому вы можете изменить функцию updateAccountAmount, чтобы получить дельту вместо конечного значения, и обновить сумму, если конечная сумма> = 0. Это set amount=amount+delta where amount+delta>0. Это будет как проверить, так и обновить. Вам не понадобятся циклы while, потому что, если первая операция завершится неудачно, нет смысла повторять попытку.

Вторая реализация, как она написана, неверна. Вам необходимо выполнить синхронизацию на общем объекте, соответствующем номеру учетной записи и номеру учетной записи, чтобы предотвратить одновременное обновление. Вы можете использовать HashMap, чтобы получить объект блокировки для идентификатора, а затем заблокировать этот объект. Вы должны синхронизировать доступ к хэш-карте, и как только вы получите два объекта, вы должны заблокировать идентификаторы в одном и том же порядке для всех потоков, чтобы предотвратить взаимные блокировки (что вы, кажется, уже сделали, но ваша синхронизация находится в стеке переменная, следовательно, не будет препятствовать одновременному доступу). Однако, как только вы исправите это, оно должно работать, пока у вас работает только один экземпляр этого приложения. Тогда у вас возникнет проблема работы с постоянно растущей хэш-картой.

Более простая схема - использовать блокировку. Вы можете использовать набор блокировок, общий для потоков, с целочисленными ключами, соответствующими номерам счетов. Поскольку метод блокировки синхронизирован, вам не нужно беспокоиться о блокировке orher. Затем: 

public boolean lock(Integer id1, Integer id2) {
  synchronized(lockSet) {
    if(!lockSet.contains(id1) && !lockSet.contains(id2)) {
       lockSet.put(id1)
       lockSet.put(id2)
       return true
    }
    return false
  }
}

public void unlock(Integer id1, Integer id2) {
  synchronized(lockSet) {
     lockSet.remove(id1)
     lockSet.remove(id2)
  }
}

Затем заблокируйте перед операциями базы данных и разблокируйте после них. Это предотвращает вхождение любого другого потока, работающего с этими идентификаторами учетных записей. Эта схема страдает от тех же проблем, с которыми сталкиваются другие: если первое обновление будет успешным, а второе - неудачным, вы потеряете деньги.

...