DataNucleus Memory/Cache Handling для больших обновлений/вставок

Мы запускаем приложение в контексте Spring, используя DataNucleus в качестве нашего сопоставления ORM и mysql в качестве нашей базы данных.

Наше приложение выполняет ежедневную работу по импорту некоторых данных в нашу базу данных. Размер потока данных соответствует примерно 1 миллиону строк вставки/обновления. Сначала производительность импорта была очень хорошей, но затем она ухудшалась со временем (по мере увеличения количества выполняемых запросов), и в какой-то момент приложение зависало или переставало отвечать. Нам придется дождаться завершения всей работы, прежде чем приложение снова ответит.

Это поведение очень похоже на утечку памяти для нас, и мы внимательно изучили наш код, чтобы обнаружить любую потенциальную проблему, однако проблема не исчезла. Одна интересная вещь, которую мы обнаружили из дампа кучи, заключается в том, что org.datanucleus.ExecutionContextThreadedImpl (или HashSet/HashMap) удерживает 90% нашей памяти (5 ГБ) во время импорта. (Я прикрепил скриншот дампа ниже). Мои исследования в Интернете показали, что эта ссылка является кэшем уровня 1 (не уверен, что я прав). Мой вопрос во время большого импорта, как я могу ограничить/контролировать размер кеша уровня 1. Может быть попросить DN не кэшировать во время моего импорта?

Если это не кэш L1, какова возможная причина моей проблемы с памятью?

Наш код использует транзакцию для каждой вставки, чтобы предотвратить блокировку большого фрагмента данных в базе данных. Он вызывает метод сброса каждые 2000 вставок.

В качестве временного исправления мы перенесли процесс импорта на ночь, когда никто не использует наше приложение. Очевидно, что так не может продолжаться вечно. Пожалуйста, может кто-нибудь хотя бы указать нам правильное направление, чтобы мы могли провести больше исследований и надеяться, что сможем найти исправления.

Было бы хорошо, если бы кто-то знал о расшифровке дампа кучи

Ваша помощь была бы очень признательна всем нам здесь. Большое спасибо!

https://s3-ap-southeast-1.amazonaws.com/public-external/datanucleus_heap_dump.png

https://s3-ap-southeast-1.amazonaws.com/public-external/datanucleus_dump2.png

Код ниже — вызывающая сторона этого метода не имеет транзакции. Этот метод будет обрабатывать один объект импорта за вызов, и нам нужно ежедневно обрабатывать около 100 000 таких объектов.

@Override
@PreAuthorize("(hasUserRole('ROLE_ADMIN')")
@Transactional(propagation = Propagation.REQUIRED)
public void processImport(ImportInvestorAccountUpdate account, String advisorCompanyKey) {

    ImportInvestorAccountDescriptor invAccDesc = account
            .getInvestorAccount();

    InvestorAccount invAcc = getInvestorAccountByImportDescriptor(
            invAccDesc, advisorCompanyKey);

    try {

        ParseReportingData parseReportingData = ctx
                .getBean(ParseReportingData.class);

        String baseCCY = invAcc.getBaseCurrency();
        Date valueDate = account.getValueDate();
        ArrayList<InvestorAccountInformationILAS> infoList = parseReportingData
                .getInvestorAccountInformationILAS(null, invAcc, valueDate,
                        baseCCY);

        InvestorAccountInformationILAS info = infoList.get(0);

        PositionSnapshot snapshot = new PositionSnapshot();
        ArrayList<Position> posList = new ArrayList<Position>();
        Double totalValueInBase = 0.0;
        double totalQty = 0.0;



        for (ImportPosition importPos : account.getPositions()) {
            Asset asset = getAssetByImportDescriptor(importPos
                    .getTicker());
            PositionInsurance pos = new PositionInsurance();
            pos.setAsset(asset);
            pos.setQuantity(importPos.getUnits());
            pos.setQuantityType(Position.QUANTITY_TYPE_UNITS);
            posList.add(pos);
        }

        snapshot.setPositions(posList);
        info.setHoldings(snapshot);

        log.info("persisting a new investorAccountInformation(source:"
                + invAcc.getReportSource() + ") on " + valueDate
                + " of InvestorAccount(key:" + invAcc.getKey() + ")");
        persistenceService.updateManagementEntity(invAcc);



    } catch (Exception e) {
        throw new DataImportException(invAcc == null ? null : invAcc.getKey(), advisorCompanyKey,
                e.getMessage());
    }

}