From f5595141a306feb3104b531a913f2fcba639784f Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Fri, 25 Jul 2025 15:24:14 +0300 Subject: [PATCH] Optimize memory usage --- .../credit_tracker/facade/SignalFacade.java | 81 +++++++++++++++---- .../service/MonitoringService.java | 8 +- .../credit_tracker/service/SignalService.java | 26 ++++-- .../service/signal/SignalRestService.java | 4 +- ...MonitoringSignalResponseStaxXmlWriter.java | 2 +- .../xml/stax/SignalsStaxXmlWriter.java | 2 +- .../ru/nbch/credit_tracker/utils/CTUtil.java | 11 +++ 7 files changed, 105 insertions(+), 29 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index e4139c5..f4d3d4a 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -11,7 +11,6 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.dto.UserPackage; import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; -import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject; import ru.nbch.credit_tracker.entities.user_response.active_packages.MonitoringActivePackagesResponse; import ru.nbch.credit_tracker.entities.user_response.active_packages.Package; import ru.nbch.credit_tracker.entities.user_response.active_packages.Packages; @@ -31,11 +30,16 @@ import ru.nbch.credit_tracker.utils.ValidationUtil; import ru.nbch.credit_tracker.utils.ZipUtils; import javax.xml.stream.XMLStreamException; +import java.io.BufferedInputStream; import java.io.File; import java.io.IOException; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; import java.time.LocalDateTime; import java.util.List; import java.util.Optional; +import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; @@ -81,6 +85,7 @@ public class SignalFacade { MonitoringRequest monitoringRequest; Integer packageId; ZipInputStream zipInputStream = null; + String filePath; try { zipInputStream = new ZipInputStream(file.getInputStream()); Optional xmlFileEntry = ZipUtils.extractValidEntry(zipInputStream); @@ -95,11 +100,11 @@ public class SignalFacade { zipInputStream = new ZipInputStream(file.getInputStream()); zipInputStream.getNextEntry(); - monitoringRequest = staxXmlProcessor.read(zipInputStream, MonitoringRequest.class); - } catch (Exception e) { - if (e.getCause() instanceof XMLStreamException) { - throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError); + filePath = writeTmpFile(zipInputStream); + if (filePath == null) { + throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError); } + } catch (Exception e) { if (e instanceof SignalClientException) { throw (SignalClientException) e; } @@ -118,12 +123,17 @@ public class SignalFacade { // валидация xml try { - ValidationUtil.validateMonitoringRequest(monitoringRequest); - } catch (SignalClientException e) { - if (monitoringRequest != null) { - signalService.rejectPackage(monitoringRequest); + monitoringRequest = readAndValidateMonitoringRequest(filePath); + } catch (Exception e) { + CTUtil.removeFile(filePath); + if (e.getCause() instanceof XMLStreamException) { + throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError); } - throw e; + if (e instanceof SignalClientException) { + throw (SignalClientException) e; + } + log.error(e.getMessage(), e); + throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError); } try { @@ -151,16 +161,58 @@ public class SignalFacade { } // в отдельном потоке запускаем регистрацию субъектов, поиск фидов и расчет флагов просрочек - registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), monitoringRequest.getSubjects().getSubject()); + registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), filePath); return ResponseEntity.ok().body(new MonitoringSetupResponse()); } - private void registerFidsAndDebts(Integer packageId, String packageName, List subjects) { + private MonitoringRequest readAndValidateMonitoringRequest(String filePath) { + MonitoringRequest monitoringRequest; + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) { + monitoringRequest = staxXmlProcessor.read(in, MonitoringRequest.class, + request -> { + try { + ValidationUtil.validateMonitoringRequest(request); + } catch (SignalClientException e) { + if (request != null) { + signalService.rejectPackage(request); + } + request = null; + throw e; + } + request.getSubjects().getSubject().clear(); + }, 40000, "/MonitoringRequest/Subjects/Subject"); + } catch (IOException e) { + throw new RuntimeException("Error while parsing and validating monitoring request", e); + } + return monitoringRequest; + } + + private String writeTmpFile(ZipInputStream zipInputStream) { + String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator; + String filePath = folderPath + UUID.randomUUID() + "monitoring_request.xml"; + CTUtil.createFolder(folderPath); + + try (OutputStream out = Files.newOutputStream(Path.of(filePath))) { + + byte[] buffer = new byte[8 * 1024 * 1024]; + int bytesRead; + while ((bytesRead = zipInputStream.read(buffer)) > 0) { + out.write(buffer, 0, bytesRead); + } + + return filePath; + } catch (Exception e) { + log.error("Ошибка при записи файла запроса постановки на мониторинг на диск", e); + return null; + } + } + + private void registerFidsAndDebts(Integer packageId, String packageName, String filePath) { executor.execute(() -> { try { - signalService.registerSubjects(subjects, packageId); - subjects.clear(); + signalService.registerSubjects(filePath, packageId); + CTUtil.removeFile(filePath); signalService.updateSubjects(packageId, null); if (signalRestService.setupMonitoring(packageId, null)) { log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName); @@ -170,6 +222,7 @@ public class SignalFacade { signalService.updatePackageStatus(packageId, Status.ERROR); } } catch (Exception e) { + CTUtil.removeFile(filePath); signalService.updatePackageStatus(packageId, Status.ERROR); log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java index 9bb02e4..b877dbc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java @@ -141,7 +141,7 @@ public class MonitoringService { MonitoringSignalResponseStaxXmlWriter staxXmlWriter = new MonitoringSignalResponseStaxXmlWriter(config.getEncoding()); staxXmlWriter.writeWithStax(responsePath, response, writer -> { - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(reportPath)), 4 * 1024 * 1024)) { + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(reportPath)), 8 * 1024 * 1024)) { staxXmlProcessor.read(in, Persons.class, personsData -> { // как только считали пачку Persons сразу записали ее в результирующий файл List subjectsBatch = processPersons(personsData.getPersons()); @@ -211,13 +211,13 @@ public class MonitoringService { folder.mkdir(); } - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 4 * 1024 * 1024); + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024); BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath + fileName + ".zip")); ZipOutputStream zos = new ZipOutputStream(bos)) { ZipEntry zipEntry = new ZipEntry(fileName + ".xml"); zos.putNextEntry(zipEntry); - byte[] buffer = new byte[65536]; + byte[] buffer = new byte[8 * 1024 * 1024]; int bytesRead; while ((bytesRead = in.read(buffer)) > 0) { zos.write(buffer, 0, bytesRead); @@ -243,7 +243,7 @@ public class MonitoringService { log.info("Updating signal hits for {}", userPackage.getPackageName()); LocalDateTime responseSentAt = LocalDateTime.now(); // Считываем результирующий файл пачками и обновляем таблицу hits - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 4 * 1024 * 1024)) { + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024)) { staxXmlProcessor.read(in, MonitoringSignalResponse.class, response -> { // как только считали пачку Subjects сразу обновлям таблицу hits List hits = new ArrayList<>(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index 7469e56..86ed695 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -11,7 +11,6 @@ import ru.nbch.credit_tracker.entities.external_request.signals.Auth; import ru.nbch.credit_tracker.entities.external_request.signals.Person; import ru.nbch.credit_tracker.entities.external_request.signals.Signals; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; -import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.service.indic.FlagsService; import ru.nbch.credit_tracker.service.indic.PersonService; @@ -20,9 +19,14 @@ import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageService; import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.CTUtil; +import java.io.BufferedInputStream; import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.time.LocalDateTime; import java.util.*; import java.util.function.Consumer; @@ -39,6 +43,7 @@ public class SignalService { private final PhoneService phoneService; private final FlagsService flagsService; private final PersonService personService; + private final StaxXmlProcessor staxXmlProcessor; private final int batchSize = 10000; @@ -48,7 +53,8 @@ public class SignalService { PhoneFidMapService phoneFidMapService, PhoneService phoneService, FlagsService flagsService, - PersonService personService) { + PersonService personService, + StaxXmlProcessor staxXmlProcessor) { this.config = config; this.hitsService = hitsService; this.userPackageService = userPackageService; @@ -56,6 +62,7 @@ public class SignalService { this.phoneService = phoneService; this.flagsService = flagsService; this.personService = personService; + this.staxXmlProcessor = staxXmlProcessor; } public boolean isPackageNotUnique(MonitoringRequest request) { @@ -72,12 +79,17 @@ public class SignalService { return userPackageService.insertPackage(userPackage); } - public void registerSubjects(List subjects, Integer packageId) { + public void registerSubjects(String filePath, Integer packageId) { log.info("Start registering subjects. PackageId: {}", packageId); - for (int i = 0; i < subjects.size(); i += batchSize) { - int end = Math.min(i + batchSize, subjects.size()); - phoneFidMapService.registerSubjects(subjects.subList(i, end), packageId); - log.trace("Registered subject's batch from: {} to: {}. Total count: {}. PackageId: {}", i, end, subjects.size(), packageId); + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) { + staxXmlProcessor.read(in, MonitoringRequest.class, + request -> { + phoneFidMapService.registerSubjects(request.getSubjects().getSubject(), packageId); + log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId); + request.getSubjects().getSubject().clear(); + }, batchSize, "/MonitoringRequest/Subjects/Subject"); + } catch (IOException e) { + throw new RuntimeException("Error while parsing monitoring request", e); } log.info("Subjects were registered. PackageId: {}", packageId); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java index 56b5f23..908c229 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java @@ -197,7 +197,7 @@ public class SignalRestService { try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(response.getBody())); OutputStream out = Files.newOutputStream(Path.of(filePath))) { - byte[] buffer = new byte[65536]; + byte[] buffer = new byte[8 * 1024 * 1024]; int bytesRead; while ((bytesRead = in.read(buffer)) > 0) { out.write(buffer, 0, bytesRead); @@ -229,7 +229,7 @@ public class SignalRestService { OutputStream out = Files.newOutputStream(Path.of(path + ".gz")); GZIPOutputStream gzipOut = new GZIPOutputStream(out)) { - byte[] buffer = new byte[65536]; + byte[] buffer = new byte[8 * 1024 * 1024]; int bytesRead; while ((bytesRead = in.read(buffer)) > 0) { gzipOut.write(buffer, 0, bytesRead); diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/MonitoringSignalResponseStaxXmlWriter.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/MonitoringSignalResponseStaxXmlWriter.java index f3eed2f..3cd9343 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/MonitoringSignalResponseStaxXmlWriter.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/MonitoringSignalResponseStaxXmlWriter.java @@ -22,7 +22,7 @@ public class MonitoringSignalResponseStaxXmlWriter extends StaxXmlWriter loader) { - try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 4 * 1024 * 1024)) { + try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 8 * 1024 * 1024)) { XMLOutputFactory output = XMLOutputFactory.newInstance(); XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance(); XMLEventWriter writer = output.createXMLEventWriter(bos, charsetName); diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java index 07ef367..18d96c9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java @@ -20,7 +20,7 @@ public class SignalsStaxXmlWriter extends StaxXmlWriter { } public void writeWithStax(String filePath, Signals signals, LoaderWriter loader) { - try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 4 * 1024 * 1024)) { + try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 8 * 1024 * 1024)) { XMLOutputFactory output = XMLOutputFactory.newInstance(); XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance(); XMLEventWriter writer = output.createXMLEventWriter(bos, charsetName); diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index b4f0f99..0383d4d 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -90,4 +90,15 @@ public class CTUtil { } } } + + public static void removeFile(String filePath) { + try { + Path path = Path.of(filePath); + if (Files.exists(path)) { + Files.delete(path); + } + } catch (IOException e) { + log.warn("File {} hasn't been deleted", filePath); + } + } }